Add data to the replication backlog. * This function also increments the global replication offset stored at * g_pserver->master_repl_offset, because there is no case where we want to feed * the backlog without incrementing the offset. */
| 357 | * g_pserver->master_repl_offset, because there is no case where we want to feed |
| 358 | * the backlog without incrementing the offset. */ |
| 359 | void feedReplicationBacklog(const void *ptr, size_t len) { |
| 360 | serverAssert(GlobalLocksAcquired()); |
| 361 | const unsigned char *p = (const unsigned char*)ptr; |
| 362 | std::unique_lock<fastlock> repl_backlog_lock(g_pserver->repl_backlog_lock, std::defer_lock); |
| 363 | |
| 364 | if (g_pserver->repl_batch_idxStart >= 0) { |
| 365 | /* We are lower bounded by the lowest replica offset, or the batch offset start if not applicable */ |
| 366 | long long lower_bound = g_pserver->repl_lowest_off.load(std::memory_order_seq_cst); |
| 367 | if (lower_bound == -1) |
| 368 | lower_bound = g_pserver->repl_batch_offStart; |
| 369 | long long minimumsize = g_pserver->master_repl_offset + len - lower_bound + 1; |
| 370 | |
| 371 | if (minimumsize > g_pserver->repl_backlog_size) { |
| 372 | listIter li; |
| 373 | listNode *ln; |
| 374 | listRewind(g_pserver->slaves, &li); |
| 375 | long long maxClientBuffer = (long long)cserver.client_obuf_limits[CLIENT_TYPE_SLAVE].hard_limit_bytes; |
| 376 | if (maxClientBuffer <= 0) |
| 377 | maxClientBuffer = LLONG_MAX; // infinite essentially |
| 378 | if (cserver.repl_backlog_disk_size) |
| 379 | maxClientBuffer = std::max(g_pserver->repl_backlog_size, cserver.repl_backlog_disk_size); |
| 380 | long long min_offset = LLONG_MAX; |
| 381 | int listening_replicas = 0; |
| 382 | while ((ln = listNext(&li))) { |
| 383 | client *replica = (client*)listNodeValue(ln); |
| 384 | if (!canFeedReplicaReplBuffer(replica)) continue; |
| 385 | if (replica->flags & CLIENT_CLOSE_ASAP) continue; |
| 386 | |
| 387 | std::unique_lock<fastlock> ul(replica->lock); |
| 388 | |
| 389 | // Would this client overflow? If so close it |
| 390 | long long neededBuffer = g_pserver->master_repl_offset + len - replica->repl_curr_off + 1; |
| 391 | if (neededBuffer > maxClientBuffer) { |
| 392 | |
| 393 | sds clientInfo = catClientInfoString(sdsempty(),replica); |
| 394 | freeClientAsync(replica); |
| 395 | serverLog(LL_WARNING,"Client %s scheduled to be closed ASAP due to exceeding output buffer hard limit.", clientInfo); |
| 396 | sdsfree(clientInfo); |
| 397 | continue; |
| 398 | } |
| 399 | min_offset = std::min(min_offset, replica->repl_curr_off); |
| 400 | ++listening_replicas; |
| 401 | } |
| 402 | |
| 403 | if (min_offset == LLONG_MAX) { |
| 404 | min_offset = g_pserver->repl_batch_offStart; |
| 405 | g_pserver->repl_lowest_off = -1; |
| 406 | } else { |
| 407 | g_pserver->repl_lowest_off = min_offset; |
| 408 | } |
| 409 | |
| 410 | minimumsize = g_pserver->master_repl_offset + len - min_offset + 1; |
| 411 | serverAssert(listening_replicas == 0 || minimumsize <= maxClientBuffer); |
| 412 | |
| 413 | if (minimumsize > g_pserver->repl_backlog_size && listening_replicas) { |
| 414 | // This is an emergency overflow, we better resize to fit |
| 415 | long long newsize = std::max(g_pserver->repl_backlog_size*2, minimumsize); |
| 416 | serverLog(LL_WARNING, "Replication backlog is too small, resizing from %lld to %lld bytes", g_pserver->repl_backlog_size, newsize); |
no test coverage detected