Repository: valkey-io/valkey
File: src/rio.c (C)

Code before:
    while (len) {
        size_t count = len < RIO_CONNSET_WRITE_MAX_CHUNK_SIZE ? len : RIO_CONNSET_WRITE_MAX_CHUNK_SIZE;
        int broken = 0;
        for (j = 0; j < r->io.connset.numconns; j++) {
            if (r->io.connset.state[j] != 0) {
                /* Skip FDs already in error. */
                broken++;
                continue;
            }

            /* Make sure to write 'count' bytes to the socket regardless
             * of short writes. */
            size_t nwritten = 0;
            while (nwritten != count) {
                retval = connWrite(r->io.connset.conns[j], p + nwritten, count - nwritten);
                if (retval <= 0) {
                    /* With blocking sockets, which is the sole user of this
                     * rio target, EWOULDBLOCK is returned only because of
                     * the SO_SNDTIMEO socket option, so we translate the error
                     * into one more recognizable by the user. */
                    if (retval == -1 && errno == EWOULDBLOCK) errno = ETIMEDOUT;
                    break;
                }
                nwritten += retval;
                if (server.in_fork_child) {
                    server.stat_net_repl_output_bytes = retval;
                }
            }

            if (nwritten != count) {
                /* Mark this FD as broken. */
                r->io.connset.state[j] = errno;
                if (r->io.connset.state[j] == 0) r->io.connset.state[j] = EIO;
            }
        }
        if (broken == r->io.connset.numconns) return 0; /* All the FDs in error. */
        p += count;
        len -= count;
        r->io.connset.pos += count;
    }

    if (doflush) sdsclear(r->io.connset.buf);
    return 1;
}

/* Returns 1 or 0 for success/failure. */
static size_t rioConnsetRead(rio *r, void *buf, size_t len) {
    UNUSED(r);
    UNUSED(buf);
    UNUSED(len);
    return 0; /* Error, this target does not support reading. */

Review comment (suggestion):
Instead of using the stat, and forcing rio to be coupled to this implementation detail - can we just report the processed_bytes already tracked in `struct rio`?

https://github.com/valkey-io/valkey/blob/9b11d3d9edf6b1f1f18976305c0923c5aad60297/src/rio.h#L68-L69

Code after:
    while (len) {
        size_t count = len < RIO_CONNSET_WRITE_MAX_CHUNK_SIZE ? len : RIO_CONNSET_WRITE_MAX_CHUNK_SIZE;
        int broken = 0;
        for (j = 0; j < r->io.connset.numconns; j++) {
            if (r->io.connset.state[j] != 0) {
                /* Skip FDs already in error. */
                broken++;
                continue;
            }

            /* Make sure to write 'count' bytes to the socket regardless
             * of short writes. */
            size_t nwritten = 0;
            while (nwritten != count) {
                retval = connWrite(r->io.connset.conns[j], p + nwritten, count - nwritten);
                if (retval <= 0) {
                    /* With blocking sockets, which is the sole user of this
                     * rio target, EWOULDBLOCK is returned only because of
                     * the SO_SNDTIMEO socket option, so we translate the error
                     * into one more recognizable by the user. */
                    if (retval == -1 && errno == EWOULDBLOCK) errno = ETIMEDOUT;
                    break;
                }
                nwritten += retval;
            }

            if (nwritten != count) {
                /* Mark this FD as broken. */
                r->io.connset.state[j] = errno;
                if (r->io.connset.state[j] == 0) r->io.connset.state[j] = EIO;
            }
        }
        if (broken == r->io.connset.numconns) return 0; /* All the FDs in error. */
        p += count;
        len -= count;
        r->io.connset.pos += count;
    }

    if (doflush) sdsclear(r->io.connset.buf);
    return 1;
}

/* Returns 1 or 0 for success/failure. */
static size_t rioConnsetRead(rio *r, void *buf, size_t len) {
    UNUSED(r);
    UNUSED(buf);
    UNUSED(len);
    return 0; /* Error, this target does not support reading. */
}

/* Returns read/write position in file. */