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
|
#ifndef broker_RateFlowcontrol_h
#define broker_RateFlowcontrol_h
/*
*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*
*/
#include "qpid/sys/Time.h"
#include "qpid/sys/IntegerTypes.h"
#include <algorithm>
namespace qpid {
namespace broker {
// Class to keep track of issuing flow control to make sure that the peer doesn't exceed
// a given message rate
//
// Create the object with the target rate
// Then call sendCredit() whenever credit is issued to the peer
// Call receivedMessage() whenever a message is received, it returns the credit to issue.
//
// sentCredit() be sensibly called with a 0 parameter to indicate
// that we sent credit but treat it as if the value was 0 (we may do this at the start of the connection
// to allow our peer to send messages)
//
// receivedMessage() can be called with 0 to indicate that we've not received a message, but
// tell me what credit I can send.
class RateFlowcontrol {
uint32_t rate; // messages per second
uint32_t maxCredit; // max credit issued to client (issued at start)
uint32_t requestedCredit;
qpid::sys::AbsTime creditSent;
public:
RateFlowcontrol(uint32_t r) :
rate(r),
maxCredit(0),
requestedCredit(0),
creditSent(qpid::sys::FAR_FUTURE)
{}
uint32_t getRate() const {
return rate;
}
void sentCredit(const qpid::sys::AbsTime& t, uint32_t credit);
uint32_t receivedMessage(const qpid::sys::AbsTime& t, uint32_t msgs);
uint32_t availableCredit(const qpid::sys::AbsTime& t);
bool flowStopped() const;
};
inline void RateFlowcontrol::sentCredit(const qpid::sys::AbsTime& t, uint32_t credit) {
// If the client isn't currently requesting credit (ie it's not sent us anything yet) then
// this credit goes to the max credit held by the client (it can't go to reduce credit
// less than 0)
int32_t nextRequestedCredit = requestedCredit - credit;
if ( nextRequestedCredit<0 ) {
requestedCredit = 0;
maxCredit -= nextRequestedCredit;
} else {
requestedCredit = nextRequestedCredit;
}
creditSent = t;
}
inline uint32_t RateFlowcontrol::availableCredit(const qpid::sys::AbsTime& t) {
qpid::sys::Duration d(creditSent, t);
// Could be -ve before first sentCredit
int64_t toSend = std::min(rate * d / qpid::sys::TIME_SEC, static_cast<int64_t>(requestedCredit));
return toSend > 0 ? toSend : 0;
}
inline uint32_t RateFlowcontrol::receivedMessage(const qpid::sys::AbsTime& t, uint32_t msgs) {
requestedCredit +=msgs;
// Don't send credit for every message, only send if more than 0.5s since last credit or
// we've got less than .25 of the max left (heuristic)
return requestedCredit*4 >= maxCredit*3 || qpid::sys::Duration(creditSent, t) >= 500*qpid::sys::TIME_MSEC
? availableCredit(t)
: 0;
}
inline bool RateFlowcontrol::flowStopped() const {
return requestedCredit >= maxCredit;
}
}}
#endif // broker_RateFlowcontrol_h
|
