1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
/* -*- c++ -*- */
// $Id$
#if !defined (JAWS_CONCURRENCY_H)
#define JAWS_CONCURRENCY_H
#include "ace/Singleton.h"
#include "ace/Synch.h"
#include "ace/Task.h"
#include "JAWS/Export.h"
#include "JAWS/IO.h"
class JAWS_Data_Block;
class JAWS_Dispatch_Policy;
class JAWS_Concurrency_Base : public ACE_Task<ACE_MT_SYNCH>
// = TITLE
// Base class for different concurrency models
//
// = DESCRIPTION
// Provides a default implementaion of the virtual put() method
// which calls putq(), but can be overloaded to do something
// synchronously, such as call put_next().
{
public:
JAWS_Concurrency_Base (void);
~JAWS_Concurrency_Base (void);
virtual int put (ACE_Message_Block *mb, ACE_Time_Value *tv = 0);
virtual int svc (void);
virtual int svc_loop (JAWS_Data_Block *db);
// in thread pool, this is an infinite loop
// in thread per request, it is a single iteration
virtual int svc_hook (JAWS_Data_Block *db);
// does the work of following the pipeline tasks
virtual int activate_hook (void);
// callback for IO_Handler when accept completes
virtual ACE_Message_Block *singleton_mb (void);
protected:
int mb_acquired_;
ACE_Message_Block *mb_;
ACE_Thread_Mutex lock_;
};
class JAWS_Dispatcher
// = TITLE
// The class that is responsible to delivering events to the
// appropriate concurrency mechanism.
//
// = DESCRIPTION
// JAWS_IO_Handler calls into the dispatcher so that the completed
// IO can find a thread to take care of it.
{
public:
JAWS_Dispatcher (void);
int dispatch (ACE_Message_Block *mb);
JAWS_Dispatch_Policy *policy (void);
JAWS_Dispatch_Policy *policy (JAWS_Dispatch_Policy *p);
private:
JAWS_Dispatch_Policy *policy_;
};
class JAWS_Thread_Pool_Task : public JAWS_Concurrency_Base
// = TITLE
// Used to implement Thread Pool Concurrency Strategy
//
// = DESCRIPTION
// This task is created to hold a pool of threads that receive
// requests through the message queue.
{
public:
virtual int open (long flags = THR_NEW_LWP,
int nthreads = 5, int maxthreads = 20);
// Initiate the thread_pool task
private:
long flags_;
int nthreads_;
int maxthreads_;
};
class JAWS_Thread_Per_Task : public JAWS_Concurrency_Base
// = TITLE
// Used to implement Thread Per Request Concurrency Strategy
//
// = DESCRIPTION
// As each new message arrives from the queue, a new thread is
// spawned to handle it. This is done by overloading put to call
// activate.
{
public:
virtual int open (long flags = THR_NEW_LWP, int maxthreads = 20);
// Initiate the thread_per task
virtual int put (ACE_Message_Block *mb, ACE_Time_Value *tv = 0);
virtual int svc_loop (JAWS_Data_Block *db);
// a single iteration
virtual int activate_hook (void);
// callback for IO_Handler when accept completes
private:
long flags_;
int maxthreads_;
};
typedef ACE_Singleton<JAWS_Dispatcher, ACE_SYNCH_MUTEX>
JAWS_Dispatcher_Singleton;
typedef ACE_Singleton<JAWS_Thread_Pool_Task, ACE_SYNCH_MUTEX>
JAWS_Thread_Pool_Singleton;
typedef ACE_Singleton<JAWS_Thread_Per_Task, ACE_SYNCH_MUTEX>
JAWS_Thread_Per_Singleton;
#endif /* !defined (JAWS_CONCURRENCY_H) */
|
