This is an automated email from the ASF dual-hosted git repository.
yong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
The following commit(s) were added to refs/heads/master by this push:
new 7f31748 improve the throttle function (#2991)
7f31748 is described below
commit 7f317486abf6fab2bb7e05d8d9cd728c9285e773
Author: StevenLuMT <[email protected]>
AuthorDate: Wed Jan 19 10:54:14 2022 +0800
improve the throttle function (#2991)
Descriptions of the changes in this PR:
### Motivation
improve the throttle function : old pr: #2778
1. duplicate definition for replicationRateByBytes
2.make sure this update safety when different callback run averageEntrySize
updating
### Changes
1.clean code for duplicate definition for replicationRateByBytes
2.add a lock to make sure data safety for updateAverageEntrySize
---
.../client/LedgerFragmentReplicator.java | 10 ++++++---
.../bookkeeper/conf/AbstractConfiguration.java | 24 ++++++++++++++++++++++
.../bookkeeper/conf/ClientConfiguration.java | 24 ----------------------
.../bookkeeper/conf/ServerConfiguration.java | 23 ---------------------
4 files changed, 31 insertions(+), 50 deletions(-)
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerFragmentReplicator.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerFragmentReplicator.java
index 65c7101..90e2006 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerFragmentReplicator.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerFragmentReplicator.java
@@ -402,9 +402,7 @@ public class LedgerFragmentReplicator {
lh.getLastAddConfirmed(), entry.getLength(),
Unpooled.wrappedBuffer(data, 0, data.length));
if (replicationThrottle != null) {
- int toSendSize = toSend.readableBytes();
- averageEntrySize = (int) (averageEntrySize *
AVERAGE_ENTRY_SIZE_RATIO
- + (1 - AVERAGE_ENTRY_SIZE_RATIO) * toSendSize);
+ updateAverageEntrySize(toSend.readableBytes());
}
for (BookieId newBookie : newBookies) {
long startWriteEntryTime = MathUtils.nowInNano();
@@ -420,6 +418,12 @@ public class LedgerFragmentReplicator {
}, null);
}
+ /* make sure this update safety when different callback run this updating
*/
+ private synchronized void updateAverageEntrySize(int toSendSize) {
+ averageEntrySize = (int) (averageEntrySize * AVERAGE_ENTRY_SIZE_RATIO
+ + (1 - AVERAGE_ENTRY_SIZE_RATIO) * toSendSize);
+ }
+
/**
* Callback for recovery of a single ledger fragment. Once the fragment has
* had all entries replicated, update the ensemble in zookeeper. Once
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/AbstractConfiguration.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/AbstractConfiguration.java
index 3b80b52..25d41cb 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/AbstractConfiguration.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/AbstractConfiguration.java
@@ -189,6 +189,8 @@ public abstract class AbstractConfiguration<T extends
AbstractConfiguration>
// option to limit stats logging
public static final String LIMIT_STATS_LOGGING = "limitStatsLogging";
+ protected static final String REPLICATION_RATE_BY_BYTES =
"replicationRateByBytes";
+
protected AbstractConfiguration() {
super();
if (READ_SYSTEM_PROPERTIES) {
@@ -1206,6 +1208,28 @@ public abstract class AbstractConfiguration<T extends
AbstractConfiguration>
}
/**
+ * Get the bytes rate of re-replication.
+ * Default value is -1 which it means entries will replicated without any
throttling activity.
+ *
+ * @return bytes rate of re-replication.
+ */
+ public int getReplicationRateByBytes() {
+ return getInt(REPLICATION_RATE_BY_BYTES, -1);
+ }
+
+ /**
+ * Set the bytes rate of re-replication.
+ *
+ * @param rate bytes rate of re-replication.
+ *
+ * @return ClientConfiguration
+ */
+ public T setReplicationRateByBytes(int rate) {
+ this.setProperty(REPLICATION_RATE_BY_BYTES, rate);
+ return getThis();
+ }
+
+ /**
* Trickery to allow inheritance with fluent style.
*/
protected abstract T getThis();
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java
index 57f0323..9354451 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java
@@ -200,30 +200,6 @@ public class ClientConfiguration extends
AbstractConfiguration<ClientConfigurati
protected static final String
CLIENT_CONNECT_BOOKIE_UNAVAILABLE_LOG_THROTTLING =
"clientConnectBookieUnavailableLogThrottling";
- protected static final String REPLICATION_RATE_BY_BYTES =
"replicationRateByBytes";
-
- /**
- * Get the bytes rate of re-replication.
- * Default value is -1 which it means entries will replicated without any
throttling activity.
- *
- * @return bytes rate of re-replication.
- */
- public int getReplicationRateByBytes() {
- return getInt(REPLICATION_RATE_BY_BYTES, -1);
- }
-
- /**
- * Set the bytes rate of re-replication.
- *
- * @param rate bytes rate of re-replication.
- *
- * @return ClientConfiguration
- */
- public ClientConfiguration setReplicationRateByBytes(int rate) {
- this.setProperty(REPLICATION_RATE_BY_BYTES, rate);
- return this;
- }
-
/**
* Construct a default client-side configuration.
*/
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java
index 8854524..a587145 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java
@@ -213,7 +213,6 @@ public class ServerConfiguration extends
AbstractConfiguration<ServerConfigurati
"auditorMaxNumberOfConcurrentOpenLedgerOperations";
protected static final String
AUDITOR_ACQUIRE_CONCURRENT_OPEN_LEDGER_OPERATIONS_TIMEOUT_MSEC =
"auditorAcquireConcurrentOpenLedgerOperationsTimeOutMSec";
- protected static final String REPLICATION_RATE_BY_BYTES =
"replicationRateByBytes";
protected static final String IN_FLIGHT_READ_ENTRY_NUM_IN_LEDGER_CHECKER =
"inFlightReadEntryNumInLedgerChecker";
@@ -3719,16 +3718,6 @@ public class ServerConfiguration extends
AbstractConfiguration<ServerConfigurati
}
/**
- * Get the bytes rate of re-replication.
- * Default value is -1 which it means entries will replicated without any
throttling activity.
- *
- * @return bytes rate of re-replication.
- */
- public int getReplicationRateByBytes() {
- return getInt(REPLICATION_RATE_BY_BYTES, -1);
- }
-
- /**
* Get in flight read entry number when ledger checker.
* Default value is -1 which it is unlimited when ledger checker.
*
@@ -3739,18 +3728,6 @@ public class ServerConfiguration extends
AbstractConfiguration<ServerConfigurati
}
/**
- * Set the rate of re-replication.
- *
- * @param rate bytes rate of re-replication.
- *
- * @return ServerConfiguration
- */
- public ServerConfiguration setReplicationRateByBytes(int rate) {
- setProperty(REPLICATION_RATE_BY_BYTES, rate);
- return this;
- }
-
- /**
* Enabled data integrity checker.
* The data integrity checker checks that the bookie has all the entries
which
* ledger metadata asserts it has.
