mirror of
https://github.com/minio/minio
synced 2024-07-22 02:44:15 +00:00
d4b391de1b
Replace the `io.Pipe` from streamingBitrotWriter -> CreateFile with a fixed size ring buffer. This will add an output buffer for encoded shards to be written to disk - potentially via RPC. This will remove blocking when `(*streamingBitrotWriter).Write` is called, and it writes hashes and data. With current settings, the write looks like this: ``` Outbound ┌───────────────────┐ ┌────────────────┐ ┌───────────────┐ ┌────────────────┐ │ │ Parr. │ │ (http body) │ │ │ │ │ Bitrot Hash │ Write │ Pipe │ Read │ HTTP buffer │ Write (syscall) │ TCP Buffer │ │ Erasure Shard │ ──────────► │ (unbuffered) │ ────────────► │ (64K Max) │ ───────────────────► │ (4MB) │ │ │ │ │ │ (io.Copy) │ │ │ └───────────────────┘ └────────────────┘ └───────────────┘ └────────────────┘ ``` We write a Hash (32 bytes). Since the pipe is unbuffered, it will block until the 32 bytes have been delivered to the TCP buffer, and the next Read hits the Pipe. Then we write the shard data. This will typically be bigger than 64KB, so it will block until two blocks have been read from the pipe. When we insert a ring buffer: ``` Outbound ┌───────────────────┐ ┌────────────────┐ ┌───────────────┐ ┌────────────────┐ │ │ │ │ (http body) │ │ │ │ │ Bitrot Hash │ Write │ Ring Buffer │ Read │ HTTP buffer │ Write (syscall) │ TCP Buffer │ │ Erasure Shard │ ──────────► │ (2MB) │ ────────────► │ (64K Max) │ ───────────────────► │ (4MB) │ │ │ │ │ │ (io.Copy) │ │ │ └───────────────────┘ └────────────────┘ └───────────────┘ └────────────────┘ ``` The hash+shard will fit within the ring buffer, so writes will not block - but will complete after a memcopy. Reads can fill the 64KB buffer if there is data for it. If the network is congested, the ring buffer will become filled, and all syscalls will be on full buffers. Only when the ring buffer is filled will erasure coding start blocking. Since there is always "space" to write output data, we remove the parallel writing since we are always writing to memory now, and the goroutine synchronization overhead probably not worth taking. If the output were blocked in the existing, we would still wait for it to unblock in parallel write, so it would make no difference there - except now the ring buffer smoothes out the load. There are some micro-optimizations we could look at later. The biggest is that, in most cases, we could encode directly to the ring buffer - if we are not at a boundary. Also, "force filling" the Read requests (i.e., blocking until a full read can be completed) could be investigated and maybe allow concurrent memory on read and write.
111 lines
2.9 KiB
Go
111 lines
2.9 KiB
Go
// Copyright (c) 2015-2021 MinIO, Inc.
|
|
//
|
|
// This file is part of MinIO Object Storage stack
|
|
//
|
|
// This program is free software: you can redistribute it and/or modify
|
|
// it under the terms of the GNU Affero General Public License as published by
|
|
// the Free Software Foundation, either version 3 of the License, or
|
|
// (at your option) any later version.
|
|
//
|
|
// This program is distributed in the hope that it will be useful
|
|
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
// GNU Affero General Public License for more details.
|
|
//
|
|
// You should have received a copy of the GNU Affero General Public License
|
|
// along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
package cmd
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
)
|
|
|
|
// Writes to multiple writers
|
|
type multiWriter struct {
|
|
writers []io.Writer
|
|
writeQuorum int
|
|
errs []error
|
|
}
|
|
|
|
// Write writes data to writers.
|
|
func (p *multiWriter) Write(ctx context.Context, blocks [][]byte) error {
|
|
for i := range p.writers {
|
|
if p.errs[i] != nil {
|
|
continue
|
|
}
|
|
if p.writers[i] == nil {
|
|
p.errs[i] = errDiskNotFound
|
|
continue
|
|
}
|
|
var n int
|
|
n, p.errs[i] = p.writers[i].Write(blocks[i])
|
|
if p.errs[i] == nil {
|
|
if n != len(blocks[i]) {
|
|
p.errs[i] = io.ErrShortWrite
|
|
p.writers[i] = nil
|
|
}
|
|
} else {
|
|
p.writers[i] = nil
|
|
}
|
|
}
|
|
|
|
// If nilCount >= p.writeQuorum, we return nil. This is because HealFile() uses
|
|
// CreateFile with p.writeQuorum=1 to accommodate healing of single disk.
|
|
// i.e if we do no return here in such a case, reduceWriteQuorumErrs() would
|
|
// return a quorum error to HealFile().
|
|
nilCount := countErrs(p.errs, nil)
|
|
if nilCount >= p.writeQuorum {
|
|
return nil
|
|
}
|
|
|
|
writeErr := reduceWriteQuorumErrs(ctx, p.errs, objectOpIgnoredErrs, p.writeQuorum)
|
|
return fmt.Errorf("%w (offline-disks=%d/%d)", writeErr, countErrs(p.errs, errDiskNotFound), len(p.writers))
|
|
}
|
|
|
|
// Encode reads from the reader, erasure-encodes the data and writes to the writers.
|
|
func (e *Erasure) Encode(ctx context.Context, src io.Reader, writers []io.Writer, buf []byte, quorum int) (total int64, err error) {
|
|
writer := &multiWriter{
|
|
writers: writers,
|
|
writeQuorum: quorum,
|
|
errs: make([]error, len(writers)),
|
|
}
|
|
|
|
for {
|
|
var blocks [][]byte
|
|
n, err := io.ReadFull(src, buf)
|
|
if err != nil {
|
|
if !IsErrIgnored(err, []error{
|
|
io.EOF,
|
|
io.ErrUnexpectedEOF,
|
|
}...) {
|
|
return 0, err
|
|
}
|
|
}
|
|
|
|
eof := err == io.EOF || err == io.ErrUnexpectedEOF
|
|
if n == 0 && total != 0 {
|
|
// Reached EOF, nothing more to be done.
|
|
break
|
|
}
|
|
|
|
// We take care of the situation where if n == 0 and total == 0 by creating empty data and parity files.
|
|
blocks, err = e.EncodeData(ctx, buf[:n])
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
if err = writer.Write(ctx, blocks); err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
total += int64(n)
|
|
if eof {
|
|
break
|
|
}
|
|
}
|
|
return total, nil
|
|
}
|