← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to plugin/pbms/src/cslib/CSThread.h

  • Committer: Brian Aker
  • Date: 2010-10-08 16:11:47 UTC
  • mto: This revision was merged to the branch mainline in revision 1834.
  • Revision ID: brian@tangent.org-20101008161147-o3us2a8itszsaycx
Typdef our lock type.

Show diffs side-by-side

added added

removed removed

Lines of Context:
plugin/pbms/src/cslib/CSThread.h
 
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 (H&G2JCtL)
 
20
 * Continued development: Barry Leslie
 
21
 *
 
22
 * 2007-05-20
 
23
 *
 
24
 * CORE SYSTEM:
 
25
 * A independently running thread.
 
26
 *
 
27
 */
 
28
 
 
29
#ifndef __CSTHREAD_H__
 
30
#define __CSTHREAD_H__
 
31
 
 
32
#include <pthread.h>
 
33
#include <setjmp.h>
 
34
 
 
35
#include "CSDefs.h"
 
36
#include "CSMutex.h"
 
37
#include "CSException.h"
 
38
#include "CSStorage.h"
 
39
 
 
40
using namespace std;
 
41
 
 
42
#define CS_THREAD_TYPE                          int
 
43
 
 
44
/* Types of threads: */
 
45
#define CS_ANY_THREAD                           0
 
46
#define CS_THREAD                                       1
 
47
 
 
48
typedef struct CSCallStack {
 
49
        const char*     cs_func;
 
50
        const char*     cs_file;
 
51
        int                     cs_line;
 
52
} CSCallStack, *CSCallStackPtr;
 
53
 
 
54
/* 
 
55
 * The release stack contains objects that need to be
 
56
 * released when an exception occurs.
 
57
 */
 
58
#define CS_RELEASE_OBJECT               1
 
59
#define CS_RELEASE_MUTEX                2
 
60
#define CS_RELEASE_POOLED               3
 
61
#define CS_RELEASE_MEM                  4
 
62
 
 
63
typedef struct CSRelease {
 
64
        int                                             r_type;
 
65
        union {
 
66
                CSObject                        *r_object;                                      /* The object to be released. */
 
67
                CSMutex                         *r_mutex;                                       /* The mutex to be unlocked! */
 
68
                CSPooled                        *r_pooled;
 
69
                void                            *r_mem;
 
70
        } x;
 
71
} CSReleaseRec, *CSReleasePtr;
 
72
 
 
73
typedef struct CSJumpBuf {
 
74
        CSReleasePtr                    jb_res_top;
 
75
        int                                             jb_call_top;
 
76
        jmp_buf                                 jb_buffer;
 
77
} CSJumpBufRec, *CSJumpBufPtr;
 
78
 
 
79
class CSThreadList: public CSLinkedList, public CSMutex {
 
80
public:
 
81
        CSThreadList():
 
82
                CSLinkedList(),
 
83
                CSMutex()
 
84
        {
 
85
        }
 
86
 
 
87
        virtual ~CSThreadList() {
 
88
                stopAllThreads();
 
89
        }
 
90
 
 
91
        /**
 
92
         * Send the given signal to all threads, except to self!
 
93
         */
 
94
        void signalAllThreads(int sig);
 
95
 
 
96
        void quitAllThreads();
 
97
 
 
98
        void stopAllThreads();
 
99
};
 
100
 
 
101
typedef void *(*ThreadRunFunc)();
 
102
 
 
103
class CSThread : public CSRefObject {
 
104
public:
 
105
        /* The name of the thread. */
 
106
        CSString                *threadName;
 
107
        CSThreadList    *myThreadList;                          /* The thread list that this thread belongs to. */
 
108
 
 
109
        /* If this value is non-zero, this signal is pending and
 
110
         * must be thrown.
 
111
         *
 
112
         * SIGTERM, SIGQUIT - Means the thread has been terminated.
 
113
         * SIGINT - Means the thread has been interrupted.
 
114
         *
 
115
         * When a signal is throw it clears this value. This includes
 
116
         * the case when system calls return error due to interrupt.
 
117
         */
 
118
        int                             signalPending;
 
119
        bool                    ignoreSignals;
 
120
 
 
121
        /* Set to true once the thread is running (never reset!). */
 
122
        bool                    isRunning;
 
123
 
 
124
        /* Set to true when the thread must quit (never reset!): */
 
125
        bool                    myMustQuit;     
 
126
        
 
127
        /* Set to true when this tread was initialized through the internal PBMS api. */
 
128
        /* When this is the case than it must only be freed via the API as well. */
 
129
        bool                    pbms_api_owner;
 
130
 
 
131
        CSException             myException;
 
132
 
 
133
        /* Transaction references. */
 
134
#ifdef DRIZZLED
 
135
        CSSortedList    mySavePoints;
 
136
#endif
 
137
        uint32_t                        myTID;                  // Current transaction ID
 
138
        uint32_t                        myTransRef;             // Reference to the current transaction cache index
 
139
        bool                    myIsAutoCommit; // Is the current transaction in auto commit mode.
 
140
        uint32_t                        myCacheVersion; // The last transaction cache version checked. Used during overflow.
 
141
        bool                    myStartTxn;             // A flag to indicate the start of a new transaction.
 
142
        uint32_t                        myStmtCount;    // Counts the number of statements in the current transaction.
 
143
        uint32_t                        myStartStmt;    // The myStmtCount at the start of the last logical statement. (An update is 2 statements but only 1 logical statement.)
 
144
        void                    *myInfo;                // A place to hang some info. (Be carefull with this!)
 
145
        
 
146
        /* The call stack */
 
147
        int                             callTop;
 
148
        CSCallStack             callStack[CS_CALL_STACK_SIZE];
 
149
 
 
150
        /* The long jump stack: */
 
151
        int                             jumpDepth;                                                      /* The current jump depth */
 
152
        CSJumpBufRec    jumpEnv[CS_JUMP_STACK_SIZE];            /* The process environment to be restored on exception */
 
153
 
 
154
        /* The release stack */
 
155
        CSReleasePtr    relTop;                                                         /* The top of the resource stack (reference next free space). */
 
156
        CSReleaseRec    relStack[CS_RELEASE_STACK_SIZE];        /* Temporary data to be freed if an exception occurs. */
 
157
 
 
158
        CSThread(CSThreadList *list):
 
159
                CSRefObject(),
 
160
                threadName(NULL),
 
161
                myThreadList(list),
 
162
                signalPending(0),
 
163
                ignoreSignals(false),
 
164
                isRunning(false),
 
165
                myMustQuit(false),
 
166
                pbms_api_owner(false),
 
167
                myTID(0),
 
168
                myTransRef(0),
 
169
                myIsAutoCommit(true),
 
170
                myCacheVersion(0),
 
171
                myStartTxn(true),
 
172
                myStmtCount(0),
 
173
                myStartStmt(0),
 
174
                myInfo(NULL),
 
175
                callTop(0),
 
176
                jumpDepth(0),
 
177
                relTop(relStack),
 
178
                iIsMain(false),
 
179
                iRunFunc(NULL),
 
180
                iNextLink(NULL),
 
181
                iPrevLink(NULL)
 
182
        {
 
183
        }
 
184
 
 
185
        virtual ~CSThread() {
 
186
                if (threadName)
 
187
                        threadName->release();
 
188
        }
 
189
 
 
190
    /**
 
191
     * Task to be performed by this thread.
 
192
         *
 
193
     * @exception CSSystemException thrown if thread cannot be run.
 
194
         */
 
195
        virtual void *run();
 
196
 
 
197
        /**
 
198
         * Start execution of the thread.
 
199
         *
 
200
     * @exception CSSystemException thrown if thread is invalid.
 
201
         */
 
202
        void start();
 
203
 
 
204
        /*
 
205
         * Stop execution of the thread.
 
206
         */
 
207
        virtual void stop();
 
208
 
 
209
        /**
 
210
         * Wait for this thread to die.
 
211
         *
 
212
     * @exception CSSystemException thrown if thread is invalid.
 
213
         */
 
214
        void *join();
 
215
 
 
216
        /**
 
217
         * Signal the thread. Throws CSSystemException 
 
218
     * if the thread is invalid.
 
219
         */
 
220
        void signal(unsigned int);
 
221
 
 
222
        void setSignalPending(unsigned int);
 
223
 
 
224
        /**
 
225
         * Check to see if we have been interrupted by a signal
 
226
         * (i.e. there is a signal pending).
 
227
         * This function throws CSSignalException if
 
228
         * there is a signal pending.
 
229
         */
 
230
        void interrupted() { if (signalPending) throwSignal(); }
 
231
        void throwSignal();
 
232
 
 
233
        /* Log the stack to the specified depth along with the message. */
 
234
        void logStack(int depth, const char *msg);
 
235
 
 
236
        /* Log the exception, and the current stack. */
 
237
        void logException();
 
238
        
 
239
        /* Log the exception, and the current stack. */
 
240
        void logMessage();
 
241
        
 
242
        /*
 
243
         * Return true if this is the main thread.
 
244
         */
 
245
        bool isMain();
 
246
 
 
247
        /*
 
248
         * Throwing exceptions:
 
249
         */
 
250
        void releaseObjects(CSReleasePtr top);
 
251
        void throwException();
 
252
        void caught();
 
253
        bool isMe(CSThread *me) { return (me->iThread == iThread);}
 
254
        /* Make this object linkable: */
 
255
        virtual CSObject *getNextLink() { return iNextLink; }
 
256
        virtual CSObject *getPrevLink() { return iPrevLink; }
 
257
        virtual void setNextLink(CSObject *link) { iNextLink = link; }
 
258
        virtual void setPrevLink(CSObject *link) { iPrevLink = link; }
 
259
 
 
260
        friend class CSDaemon;
 
261
 
 
262
private:
 
263
        pthread_t               iThread;
 
264
        bool                    iIsMain;
 
265
        ThreadRunFunc   iRunFunc;
 
266
        CSObject                *iNextLink;
 
267
        CSObject                *iPrevLink;
 
268
 
 
269
        void addToList();
 
270
        void removeFromList();
 
271
 
 
272
public:
 
273
        /* Each thread stores is thread object in this key: */
 
274
        static pthread_key_t sThreadKey;
 
275
 
 
276
   /**
 
277
     * Put the currently executing thread to sleep for a given amount of
 
278
     * time.
 
279
     *
 
280
     * @param timeout maximum amount of time (milliseconds) this method could block
 
281
     *
 
282
     * @exception TDInterruptedException thrown if the threads sleep is interrupted
 
283
     *            before <i>timeout</i> milliseconds expire.
 
284
     */
 
285
        static void sleep(unsigned long timeout);
 
286
 
 
287
        /* Do static initialization and de-initialization. */
 
288
        static bool isUp;
 
289
        static bool startUp();
 
290
        static void shutDown();
 
291
 
 
292
        /* Attach and detach an already running thread: */
 
293
        static bool attach(CSThread *thread);
 
294
        static void detach(CSThread *thread);
 
295
 
 
296
        /**
 
297
         * Return the thread object of the current
 
298
         * thread.
 
299
         */
 
300
        static CSThread *getSelf();
 
301
        static bool setSelf(CSThread *self);
 
302
 
 
303
        static CSThread *newCSThread();
 
304
        static CSThread *newThread(CSString *name, ThreadRunFunc run_func, CSThreadList *list);
 
305
 
 
306
        /* called for a newly created thread. */
 
307
        static void *dispatch(void *arg);
 
308
 
 
309
};
 
310
 
 
311
class CSDaemon : public CSThread, public CSSync {
 
312
public:
 
313
        time_t                  myWaitTime;                                     /* Wait time in milli-seconds */
 
314
 
 
315
        CSDaemon(time_t wait_time, CSThreadList *list);
 
316
        CSDaemon(CSThreadList *list);
 
317
        virtual ~CSDaemon() { }
 
318
 
 
319
        virtual void *run();
 
320
 
 
321
        virtual bool initializeWork() { return true; };
 
322
 
 
323
        virtual bool doWork();
 
324
 
 
325
        virtual void *completeWork() { return NULL; };
 
326
 
 
327
        virtual bool handleException();
 
328
 
 
329
        virtual void stop();
 
330
 
 
331
        void wakeup();
 
332
 
 
333
        void suspend();
 
334
 
 
335
        bool isSuspend() { return (iSuspendCount != 0);} // Don't use iSuspended, we are interested in if suspend() was called.
 
336
 
 
337
        void resume();
 
338
 
 
339
        virtual void returnToPool() {
 
340
                resume();
 
341
                release();
 
342
        }
 
343
 
 
344
        void suspended();
 
345
 
 
346
        void suspendedWait();
 
347
 
 
348
        void suspendedWait(time_t milli_sec);
 
349
 
 
350
private:
 
351
        bool                    iSuspended;
 
352
        uint32_t                        iSuspendCount;
 
353
};
 
354
 
 
355
#endif