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
|
/*
* SRT - Secure, Reliable, Transport
* Copyright (c) 2021 Haivision Systems Inc.
*
* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/.
*
*/
/*****************************************************************************
Written by
Haivision Systems Inc.
*****************************************************************************/
#include "platform_sys.h"
#include <algorithm>
#include <sstream>
#include "group_backup.h"
namespace srt
{
namespace groups
{
using namespace std;
using namespace srt_logging;
const char* stateToStr(BackupMemberState state)
{
switch (state)
{
case srt::groups::BKUPST_UNKNOWN:
return "UNKNOWN";
case srt::groups::BKUPST_PENDING:
return "PENDING";
case srt::groups::BKUPST_STANDBY:
return "STANDBY";
case srt::groups::BKUPST_ACTIVE_FRESH:
return "ACTIVE_FRESH";
case srt::groups::BKUPST_ACTIVE_STABLE:
return "ACTIVE_STABLE";
case srt::groups::BKUPST_ACTIVE_UNSTABLE:
return "ACTIVE_UNSTABLE";
case srt::groups::BKUPST_ACTIVE_UNSTABLE_WARY:
return "ACTIVE_UNSTABLE_WARY";
case srt::groups::BKUPST_BROKEN:
return "BROKEN";
default:
break;
}
return "WRONG_STATE";
}
/// @brief Compares group members by their weight (higher weight comes first), then state.
/// Higher weight comes first, same weight: stable, then fresh active.
struct FCompareByWeight
{
/// @returns true if the first argument is less than (i.e. is ordered before) the second.
bool operator()(const BackupMemberStateEntry& a, const BackupMemberStateEntry& b)
{
if (a.pSocketData != NULL && b.pSocketData != NULL
&& (a.pSocketData->weight != b.pSocketData->weight))
return a.pSocketData->weight > b.pSocketData->weight;
if (a.state != b.state)
{
SRT_STATIC_ASSERT(BKUPST_ACTIVE_STABLE > BKUPST_ACTIVE_FRESH, "Wrong ordering");
return a.state > b.state;
}
// the order does not matter, but comparator must return a different value for not equal a and b
return a.socketID < b.socketID;
}
};
void SendBackupCtx::recordMemberState(SocketData* pSockData, BackupMemberState st)
{
m_memberStates.push_back(BackupMemberStateEntry(pSockData, st));
++m_stateCounter[st];
if (st == BKUPST_STANDBY)
{
m_standbyMaxWeight = max(m_standbyMaxWeight, pSockData->weight);
}
else if (isStateActive(st))
{
m_activeMaxWeight = max(m_activeMaxWeight, pSockData->weight);
}
}
void SendBackupCtx::updateMemberState(const SocketData* pSockData, BackupMemberState st)
{
typedef vector<BackupMemberStateEntry>::iterator iter_t;
for (iter_t i = m_memberStates.begin(); i != m_memberStates.end(); ++i)
{
if (i->pSocketData == NULL)
continue;
if (i->pSocketData != pSockData)
continue;
if (i->state == st)
return;
--m_stateCounter[i->state];
++m_stateCounter[st];
i->state = st;
return;
}
LOGC(gslog.Error,
log << "IPE: SendBackupCtx::updateMemberState failed to locate member");
}
void SendBackupCtx::sortByWeightAndState()
{
sort(m_memberStates.begin(), m_memberStates.end(), FCompareByWeight());
}
BackupMemberState SendBackupCtx::getMemberState(const SocketData* pSockData) const
{
typedef vector<BackupMemberStateEntry>::const_iterator const_iter_t;
for (const_iter_t i = m_memberStates.begin(); i != m_memberStates.end(); ++i)
{
if (i->pSocketData != pSockData)
continue;
return i->state;
}
// The entry was not found
// TODO: Maybe throw an exception here?
return BKUPST_UNKNOWN;
}
unsigned SendBackupCtx::countMembersByState(BackupMemberState st) const
{
return m_stateCounter[st];
}
std::string SendBackupCtx::printMembers() const
{
stringstream ss;
typedef vector<BackupMemberStateEntry>::const_iterator const_iter_t;
for (const_iter_t i = m_memberStates.begin(); i != m_memberStates.end(); ++i)
{
ss << "@" << i->socketID << " w " << i->pSocketData->weight << " state " << stateToStr(i->state) << ", ";
}
return ss.str();
}
} // namespace groups
} // namespace srt
|
