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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
|
// cvs-id : $Id$
#ifndef REPLICATION_MANAGER_H
#define REPLICATION_MANAGER_H
#include "LWFTS.h"
#include "ace/Hash_Map_Manager_T.h"
#include "ace/Unbounded_Queue.h"
#include "ace/Unbounded_Set.h"
#include "ace/Condition_T.h"
#include "ace/Containers_T.h"
#include "Timer.h"
#include "ace/Recursive_Thread_Mutex.h"
#include <list>
#include <queue>
typedef enum { PRIMARY=1, BACKUP=2 } Role;
typedef enum { PROCESS_LEVEL=1, PROCESSOR_LEVEL=2 } AlgoMode;
class Algorithm;
struct APP_INFO
{
ACE_CString object_id;
double load;
ACE_CString host_name;
ACE_CString process_id;
Role role;
CORBA::Object_var ior;
APP_INFO (void);
APP_INFO (APP_INFO const & app_info);
APP_INFO (const char *oid, double l, const char *hname,
const char *pid, Role r, CORBA::Object_ptr ref);
APP_INFO (const char *oid, const char *hname,
const char *pid, Role r);
void swap (APP_INFO & app_info);
APP_INFO & operator = (APP_INFO const & app_info);
};
bool operator == (APP_INFO const & lhs, APP_INFO const & rhs);
struct RANKED_IOR_LIST
{
bool now;
std::list<CORBA::Object_var> ior_list;
std::list<ACE_CString> host_list;
RANKED_IOR_LIST (void);
};
struct UtilRank
{
double util;
std::string host_id;
UtilRank ();
UtilRank (UtilRank const & ur);
UtilRank (double u, const char * hid);
};
bool operator < (UtilRank const & u1, UtilRank const & u2);
struct MonitorUpdate
{
typedef enum { PROC_FAIL_UPDATE, HOST_UTIL_UPDATE, RUN_NOW, APP_REG } UpdateType;
UpdateType type;
ACE_CString process_id;
ACE_CString host_id;
double value;
APP_INFO app_info;
static MonitorUpdate *
create_proc_fail_update(const char * pid);
static MonitorUpdate *
create_host_util_update(const char *hid, double value);
static MonitorUpdate *
create_run_now_update();
static MonitorUpdate *
create_app_info_update (const char *oid, double l,
const char *hname, const char *pid,
Role r, CORBA::Object_ptr ref);
};
class ReplicationManager_i : public virtual POA_ReplicationManager,
protected Timer
{
public:
ReplicationManager_i (CORBA::ORB_ptr orb, double hertz,
bool proactive = true, AlgoMode mode = PROCESS_LEVEL);
~ReplicationManager_i (void);
virtual void register_application (const char *object_id,
double load,
const char *host_name,
const char *process_id,
CORBA::Short role,
CORBA::Object_ptr server_reference);
void util_update (const char *host_id,
double util);
virtual void proc_failure (const char *process_id);
virtual RankList * register_agent (CORBA::Object_ptr agent_reference);
virtual CORBA::Object_ptr get_next (const char * object_id);
bool replica_selection_algo ();
typedef ACE_Unbounded_Set<ACE_CString> STRING_LIST;
typedef ACE_Unbounded_Set<APP_INFO> APP_SET;
typedef ACE_Hash_Map_Manager_Ex<
ACE_CString,
STRING_LIST,
ACE_Hash<ACE_CString>,
ACE_Equal_To<ACE_CString>,
ACE_Null_Mutex> STRING_TO_STRING_LIST_MAP;
typedef ACE_Hash_Map_Manager_Ex<
ACE_CString,
ACE_CString,
ACE_Hash<ACE_CString>,
ACE_Equal_To<ACE_CString>,
ACE_Null_Mutex> STRING_TO_STRING_MAP;
typedef ACE_Hash_Map_Manager_Ex<
ACE_CString,
double,
ACE_Hash<ACE_CString>,
ACE_Equal_To<ACE_CString>,
ACE_Null_Mutex> STRING_TO_DOUBLE_MAP;
typedef ACE_Hash_Map_Manager_Ex<
ACE_CString,
APP_SET,
ACE_Hash<ACE_CString>,
ACE_Equal_To<ACE_CString>,
ACE_Null_Mutex> OBJECTID_APPSET_MAP;
typedef ACE_Hash_Map_Manager_Ex<
ACE_CString,
RANKED_IOR_LIST,
ACE_Hash<ACE_CString>,
ACE_Equal_To<ACE_CString>,
ACE_Null_Mutex> OBJECTID_RANKED_IOR_MAP;
typedef ACE_Unbounded_Set<CORBA::Object_var> AGENT_LIST;
enum { UPDATE_LIST_MAX_SIZE = 100 };
private:
CORBA::ORB_var orb_;
Algorithm * algo_thread_;
bool proactive_;
AlgoMode mode_;
ACE_DLList <MonitorUpdate> update_list_;
ACE_Thread_Mutex update_mutex_;
ACE_Condition <ACE_Thread_Mutex> update_available_;
ACE_Condition <ACE_Thread_Mutex> update_list_full_;
OBJECTID_APPSET_MAP objectid_appset_map_;
ACE_Recursive_Thread_Mutex appset_lock_;
OBJECTID_RANKED_IOR_MAP objectid_rankedior_map_;
STRING_TO_DOUBLE_MAP hostid_util_map_;
STRING_TO_DOUBLE_MAP objectid_load_map_;
STRING_TO_STRING_MAP processid_host_map_;
STRING_TO_STRING_LIST_MAP processid_backup_map_;
STRING_TO_STRING_LIST_MAP processid_primary_map_;
STRING_TO_STRING_LIST_MAP hostid_process_map_;
RankList rank_list_;
AGENT_LIST agent_list_;
ACE_Thread_Mutex rank_list_agent_list_combined_mutex_;
void update_map (const char * key_str,
const char * value_str,
STRING_TO_STRING_LIST_MAP & map);
void update_util_map (const char * key_str,
double value,
STRING_TO_DOUBLE_MAP & map);
void update_appset_map (const char * key_str,
APP_INFO const & app_info,
OBJECTID_APPSET_MAP &);
void update_ior_map (ACE_CString const & oid,
std::priority_queue<UtilRank> const & rank_list);
void update_proc_host_map (const char *pid,
const char * hid,
STRING_TO_STRING_MAP & map);
std::priority_queue <UtilRank>
util_sorted_host_list (ACE_CString const & oid,
STRING_LIST const & host_list,
STRING_TO_DOUBLE_MAP const & hu_map);
virtual int pulse();
void move_update_list (ACE_DLList<MonitorUpdate> & source,
ACE_DLList<MonitorUpdate> & dest);
bool process_updates(ACE_DLList<MonitorUpdate> & update_list);
void remove_process(ACE_CString const & pid,
STRING_TO_STRING_LIST_MAP & map,
ACE_CString const & host,
Role role);
void send_rank_list ();
void build_rank_list ();
void app_reg(APP_INFO & app_info);
void static_ranklist_update (const char * object_id,
CORBA::Object_ptr ior,
Role role);
void process_proc_failure (ACE_CString const & process_id);
STRING_LIST non_primary_host_list (ACE_CString const & primary_object_id);
void replace_primary_tags (ACE_CString const & pid, ACE_CString const & host);
void remove_from_appset (ACE_CString const & host, ACE_CString const & pid,
ACE_CString const & tag, Role role);
void elevate_backup_to_primary (ACE_CString const & tag);
void replace_backup_tags (ACE_CString const & pid, ACE_CString const & host);
void copy_map (STRING_TO_DOUBLE_MAP const & source, STRING_TO_DOUBLE_MAP & dest);
void print_queue (std::priority_queue <UtilRank> queue);
// RankList * rank_list_clone ();
};
#endif /* REPLICATION_MANAGER_H */
|
