Method flushData

src/replication.cpp:1026–1071

    bool isActive() const { return !replicas.empty(); }

    void flushData() {
        if (reply == nullptr)
            return;
        size_t written = reply->used;

        for (auto replica : replicas) {
            std::unique_lock<fastlock> ul(replica->lock);
            while (checkClientOutputBufferLimits(replica)
              && (replica->flags.load(std::memory_order_relaxed) & CLIENT_CLOSE_ASAP) == 0) {
                ul.unlock();
                usleep(0);
                ul.lock();
            }
        }

        aeAcquireLock();
        for (size_t ireplica = 0; ireplica < replicas.size(); ++ireplica) {
            auto replica = replicas[ireplica];
            if (replica->flags.load(std::memory_order_relaxed) & CLIENT_CLOSE_ASAP) {
                replica->replstate = REPL_STATE_NONE;
                continue;
            }
            
            std::unique_lock<fastlock> lock(replica->lock);
            if (ireplica == (replicas.size()-1) && replica->replyAsync == nullptr) {
                replica->replyAsync = reply;
                reply = nullptr;
                if (!(replica->fPendingAsyncWrite)) clientInstallAsyncWriteHandler(replica);
            } else {
                addReplyProto(replica, reply->buf(), reply->used);
            }
        }
        ProcessPendingAsyncWrites();
        for (auto c : replicas) {
            if (c->flags & CLIENT_CLOSE_ASAP) {
                std::unique_lock<fastlock> ul(c->lock);
                c->replstate = REPL_STATE_NONE;   // otherwise the client can't be free'd
            }
        }
        replicas.erase(std::remove_if(replicas.begin(), replicas.end(), [](const client *c)->bool{ return c->flags.load(std::memory_order_relaxed) & CLIENT_CLOSE_ASAP;}), replicas.end());
        aeReleaseLock();
        if (reply != nullptr) {
            reply->used = 0;
        }
        writtenBytesTracker += written;
    }

    void addData(const char *data, unsigned long size) {
        if (reply != nullptr && (size + reply->used) > reply->size)