added mechanism for getting rid of spikes
diff --git a/contrib/adaptive-compression/adapt.c b/contrib/adaptive-compression/adapt.c
index f6bcddd..32f54e6 100644
--- a/contrib/adaptive-compression/adapt.c
+++ b/contrib/adaptive-compression/adapt.c
@@ -26,6 +26,7 @@
#define DEFAULT_COMPRESSION_LEVEL 6
#define DEFAULT_ADAPT_PARAM 0
#define MAX_COMPRESSION_LEVEL_CHANGE 2
+#define CONVERGENCE_LOWER_BOUND 3
static int g_displayLevel = DEFAULT_DISPLAY_LEVEL;
static unsigned g_compressionLevel = DEFAULT_COMPRESSION_LEVEL;
@@ -77,6 +78,7 @@
unsigned jobWriteID;
unsigned allJobsCompleted;
unsigned adaptParam;
+ unsigned convergenceCounter;
double createWaitCompressionCompletion;
double compressWaitCreateCompletion;
double compressWaitWriteCompletion;
@@ -213,6 +215,7 @@
ctx->createCompletion = 1;
ctx->writeCompletion = 1;
ctx->compressionCompletion = 1;
+ ctx->convergenceCounter = 0;
ctx->jobs = calloc(1, numJobs*sizeof(jobDescription));
@@ -323,6 +326,7 @@
double compressWaitWriteCompletion;
double writeWaitCompressionCompletion;
double const threshold = 0.00001;
+ unsigned const prevCompressionLevel = ctx->compressionLevel;
DEBUG(2, "adapting compression level %u\n", ctx->compressionLevel);
/* read and reset completion measurements */
@@ -347,7 +351,7 @@
compressWaitCreateCompletion = ctx->compressWaitCreateCompletion;
ctx->compressWaitCreateCompletion = 1;
pthread_mutex_unlock(&ctx->createCompletion_mutex.pMutex);
-
+ DEBUG(2, "convergence counter: %u\n", ctx->convergenceCounter);
/* adaptation logic */
if (1-createWaitCompressionCompletion > threshold || 1-writeWaitCompressionCompletion > threshold) {
/* create or write waiting on compression */
@@ -355,7 +359,15 @@
double const completion = MAX(createWaitCompressionCompletion, writeWaitCompressionCompletion);
unsigned const change = (unsigned)((1-completion) * MAX_COMPRESSION_LEVEL_CHANGE);
unsigned const boundChange = MIN(change, ctx->compressionLevel - 1);
- ctx->compressionLevel -= boundChange;
+ if (ctx->convergenceCounter > CONVERGENCE_LOWER_BOUND && boundChange != 0) {
+ /* reset convergence counter, might have been a spike */
+ ctx->convergenceCounter = 0;
+ }
+ else if (boundChange != 0) {
+ ctx->compressionLevel -= boundChange;
+ ctx->convergenceCounter = 1;
+ }
+
DEBUG(2, "create or write threads waiting on compression, tried to decrease compression level by %u\n\n", boundChange);
}
else if (1-compressWaitWriteCompletion > threshold) {
@@ -363,7 +375,14 @@
double const completion = compressWaitWriteCompletion;
unsigned const change = (unsigned)((1-completion) * MAX_COMPRESSION_LEVEL_CHANGE);
unsigned const boundChange = MIN(change, ZSTD_maxCLevel() - ctx->compressionLevel);
- ctx->compressionLevel += boundChange;
+ if (ctx->convergenceCounter > CONVERGENCE_LOWER_BOUND && boundChange != 0) {
+ ctx->convergenceCounter = 0;
+ }
+ else if (boundChange != 0) {
+ ctx->compressionLevel += boundChange;
+ ctx->convergenceCounter = 1;
+ }
+
DEBUG(2, "compress waiting on write, tried to increase compression level by %u\n\n", boundChange);
}
else if (1-compressWaitCreateCompletion > threshold) {
@@ -372,10 +391,21 @@
double const completion = compressWaitCreateCompletion;
unsigned const change = (unsigned)((1-completion) * MAX_COMPRESSION_LEVEL_CHANGE);
unsigned const boundChange = MIN(change, ZSTD_maxCLevel() - ctx->compressionLevel);
- ctx->compressionLevel += boundChange;
+ if (ctx->convergenceCounter > CONVERGENCE_LOWER_BOUND && boundChange != 0) {
+ ctx->convergenceCounter = 0;
+ }
+ else if (boundChange != 0) {
+ ctx->compressionLevel += boundChange;
+ ctx->convergenceCounter = 1;
+ }
+
DEBUG(2, "compression waiting on create, tried to increase compression level by %u\n\n", boundChange);
}
+ if (ctx->compressionLevel == prevCompressionLevel) {
+ ctx->convergenceCounter++;
+ }
+
if (g_forceCompressionLevel) {
ctx->compressionLevel = g_compressionLevel;
}