xio

package module
v0.4.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 1, 2025 License: MIT Imports: 9 Imported by: 0

README

xio

Go Reference Go

xio is an extended I/O library for various needs.

Features

  • A buffered pipe implementation for concurrent read/write operations.
  • In-memory storage implementations for io.ReaderAt and io.WriterAt interfaces.
  • Caching io.ReadeAt implementation.
  • Utility function to compare data from two io.Reader.
  • Additional helper utilities.

Contributing

Contributions are welcome! Please open an issue or submit a pull request if you would like to contribute code.

License

xio is licensed under the MIT License.

Feel free to reach out if you have any questions or feedback!

Documentation

Index

Examples

Constants

View Source 
const DefaultBlockSize = 1 * 1024 * 1024

DefaultBlockSize is the default block size used when block size is not specified.

Variables

View Source 
var (
	ErrCmpReadError         = errors.New("xio: compare: read error")
	ErrCmpByteCountMismatch = errors.New("xio: compare: byte count mismatch")
	ErrCmpDataMismatch      = errors.New("xio: compare: data mismatch")
)
View Source 
var ErrNoSpaceLeft = errors.New("xio: no space left")

ErrNoSpaceLeft is returned when there is no space left in the underlying storage.

Functions

func BufPipe 

func BufPipe(blockSize int, storageSize int64, storage Storage) (*BufPipeReader, *BufPipeWriter)

BufPipe creates a buffered pipe with a given block size and storage size. It was inspired by the io.Pipe function from the Go standard library.

The blockSize should be at least 512 bytes and storageSize should be at least blockSize. If blockSize is <= 0, a default value is used, DefaultBlockSize 1MiB.

If storageSize is <= 0, it is set to blockSize. If storageSize is less than blockSize then blockSize is capped to storageSize for simplicity but in terms of performance and correctness, you should provide appropriate values.

Example (Example1) 
package main

import (
	"bytes"
	"fmt"
	"io"
	"sync"

	"github.com/ozanh/xio"
)

func main() {
	const blockSize = 512
	const storageSize = 1024 * 1024

	storage := xio.NewBlockStorageBuffer(blockSize, storageSize)
	pr, pw := xio.BufPipe(blockSize, storageSize, storage)

	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()

		_, err := pw.Write([]byte("hello"))
		if err != nil {
			panic(err)
		}
		_ = pw.Close()
	}()

	buf := bytes.NewBuffer(nil)

	_, err := io.Copy(buf, pr)
	_ = pr.CloseWithError(err)

	wg.Wait()

	fmt.Printf("%s", buf.Bytes())

}

Output:
hello
Example (Example2) 
package main

import (
	"crypto/rand"
	"fmt"
	"io"

	"github.com/ozanh/xio"
)

func main() {
	const blockSize = 1024 * 1024
	const storageSize = 10 * 1024 * 1024

	storage := xio.NewBlockStorageBuffer(blockSize, storageSize)
	pr, pw := xio.BufPipe(blockSize, storageSize, storage)

	go func() {
		src := io.LimitReader(rand.Reader, storageSize)

		_, err := io.Copy(pw, src)
		_ = pw.CloseWithError(err)
	}()

	n, err := io.Copy(io.Discard, pr)
	_ = pr.CloseWithError(err)

	fmt.Printf("%d", n)

}

Output:
10485760
Example (Example3) 
package main

import (
	"context"
	"crypto/rand"
	"fmt"
	"io"
	"sync"
	"time"

	"github.com/ozanh/xio"
)

func main() {
	const blockSize = 1024 * 1024
	const storageSize = 10 * 1024 * 1024

	storage := xio.NewBlockStorageBuffer(blockSize, storageSize)
	pr, pw := xio.BufPipe(blockSize, storageSize, storage)

	var wg sync.WaitGroup
	wg.Add(1)

	go func() {
		defer wg.Done()

		src := io.LimitReader(rand.Reader, storageSize)

		_, err := io.Copy(pw, src)
		_ = pw.CloseWithError(err)
	}()

	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()

	done := make(chan struct{})

	go func() {
		select {
		case <-ctx.Done():
			_ = pw.CloseWithError(ctx.Err())
		case <-done:
		}
	}()

	n, err := io.Copy(io.Discard, pr)
	_ = pr.CloseWithError(err)

	close(done)

	wg.Wait()

	fmt.Printf("%d", n)

}

Output:
10485760
Example (Example4) 
package main

import (
	"crypto/rand"
	"fmt"
	"io"
	"os"

	"github.com/ozanh/xio"
)

func main() {
	const blockSize = 1024 * 1024
	const storageSize = 10 * 1024 * 1024

	storage, err := os.CreateTemp("", "storage")
	if err != nil {
		panic(err)
	}
	defer os.Remove(storage.Name())
	defer storage.Close()

	err = storage.Truncate(storageSize)
	if err != nil {
		panic(err)
	}

	pr, pw := xio.BufPipe(blockSize, storageSize, storage)

	go func() {
		src := io.LimitReader(rand.Reader, 40_000_000)

		_, err := io.Copy(pw, src)
		_ = pw.CloseWithError(err)
	}()

	n, err := io.Copy(io.Discard, pr)
	if err != nil {
		panic(err)
	}
	fmt.Printf("%d", n)

}

Output:
40000000

func CmpReadersData added in v0.3.0 

func CmpReadersData[L, R io.Reader](left L, right R) error

CmpReadersData compares the data read from two readers. It returns an error if the data read from the readers is not equal. CmpReadersData reads data from the readers in chunks and compares the data in the chunks. If the data read from the readers is not equal, it returns an error with the data read from the readers. The error is of type *ReadersDataCmpError.

Example 
package main

import (
	"bufio"
	"bytes"
	"crypto/rand"
	"fmt"

	"github.com/ozanh/xio"
)

func main() {
	b := make([]byte, 1000*1000)

	_, err := xio.ReadFill(rand.Reader, b)
	if err != nil {
		panic(err)
	}

	r1 := bytes.NewReader(b)
	r2 := bufio.NewReader(bytes.NewReader(b))

	err = xio.CmpReadersData(r1, r2)
	if err != nil {
		panic(err)
	}

	fmt.Println("Readers have equal data")

}

Output:
Readers have equal data

func CmpReadersDataWithBuffer added in v0.3.0 

func CmpReadersDataWithBuffer[L, R io.Reader](left L, right R, buffer []byte) error

CmpReadersDataWithBuffer is like CmpReadersData but accepts a buffer to use for reading data from the readers. If the buffer is not provided, a new buffer of size 64KB is used. Given buffer is split into two halves and used for reading data from the readers.

func ReadFill 

func ReadFill[T io.Reader](r T, buf []byte) (n int, err error)

ReadFill reads from r into buf until it is full or an error occurs. It returns the number of bytes read into buf and the error, if any. ReadFill returns io.ErrShortBuffer if len(buf) < 1. ReadFill returns io.ErrNoProgress if the reader returns 0 bytes for maxConsecutiveEmptyReads times. ReadFill panics if the reader returns a negative count from Read.

Types

type BlockStorageBuffer 

type BlockStorageBuffer struct {
	// contains filtered or unexported fields
}

BlockStorageBuffer implements the Storage interface using a slice of byte slices. It is suitable for large data storage instead of using a single byte slice. BlockStorageBuffer methods are goroutine-safe.

func NewBlockStorageBuffer 

func NewBlockStorageBuffer(blockSize, storageSize int) *BlockStorageBuffer

NewBlockStorageBuffer creates a new BlockStorageBuffer with the given block size and storage size.

func (*BlockStorageBuffer) BlockSize 

func (bs *BlockStorageBuffer) BlockSize() int

BlockSize returns the block size.

func (*BlockStorageBuffer) ReadAt 

func (bs *BlockStorageBuffer) ReadAt(p []byte, off int64) (n int, err error)

ReadAt reads len(p) bytes from the storage starting at byte offset off. It implements io.ReaderAt interface.

func (*BlockStorageBuffer) StorageSize 

func (bs *BlockStorageBuffer) StorageSize() int

StorageSize returns the storage size.

func (*BlockStorageBuffer) WriteAt 

func (bs *BlockStorageBuffer) WriteAt(p []byte, off int64) (n int, err error)

WriteAt writes len(p) bytes to the storage starting at byte offset off. It implements io.WriterAt interface.

type BufPipeReader 

type BufPipeReader struct {
	// contains filtered or unexported fields
}

BufPipeReader is the read side of a buffered pipe. Close() or CloseWithError() should be called when the reader is no longer needed.

func (*BufPipeReader) Close 

func (br *BufPipeReader) Close() error

Close closes the reader; subsequent writes to the write side of the buffered pipe will return the error io.ErrClosedPipe.

Normally, receiving io.ErrClosedPipe error in the write side means that read side is closed without an error, maybe not interested in reading anymore.

Close() always returns nil error. See CloseWithError() for closing with an error, and an example.

func (*BufPipeReader) CloseWithError 

func (br *BufPipeReader) CloseWithError(err error) error

CloseWithError closes the reader; subsequent writes to the write side of the buffered pipe will receive the provided err.

Calling CloseWithError(nil) is the same as calling Close(). CloseWithError() always returns nil error.

Example: 

	pr, pw := BufPipe(/*...*/)

 	// ...

	n, err := io.Copy(dest, pr)
 	pipeReader.CloseWithError(err)

CloseWithError(err) is more ergonomic over Close() with io.Copy().

Subsequest Read() after Close() or CloseWithError() will receive io.ErrClosedPipe.

func (*BufPipeReader) Read 

func (br *BufPipeReader) Read(p []byte) (n int, err error)

Read implements the io.Reader interface. It reads written data from the buffer, and blocks until a written block arrives from the write side, or the read/write close happens. If write side closes normally (without an error), read side will receive io.EOF after reading the buffered data.

type BufPipeWriter 

type BufPipeWriter struct {
	// contains filtered or unexported fields
}

BufPipeWriter is the write side of a buffered pipe.

func (*BufPipeWriter) Close 

func (bw *BufPipeWriter) Close() error

Close closes the writer and flushes the last block; subsequent reads from the read side of the buffered pipe will receive io.EOF until all the buffered data is read, if read side is not closed in between.

Close() must be called when the writer is no longer needed because it flushes the last block to the storage, and read side cannot stop reading until a close happens in the write or read side.

Close must happen after all the writes are done. Subsequent Write() after Close() will receive io.ErrClosedPipe error.

Close() always returns nil error.

Calling CloseWithError(nil) is the same as calling Close(). See CloseWithError() for closing with an error, and an example.

func (*BufPipeWriter) CloseWithError 

func (bw *BufPipeWriter) CloseWithError(err error) error

CloseWithError closes the writer; subsequent reads from the read side of the buffered pipe will receive this error err. If err is nil, io.EOF is used, which is the same as calling Close().

CloseWithError() with a non-nil error can be called anytime to stop read and write operations. Note that it does not flush the last block for the read side. Closing is only taken into account from the Write() before actual write to the underlying storage.

CloseWithError() always returns nil error.

Example: 

	pr, pw := BufPipe(/*...*/)

 	// ...

	n, err := io.Copy(pw, source)
 	pw.CloseWithError(err)

Subsequent Write() after Close() or CloseWithError() will receive io.ErrClosedPipe.

func (*BufPipeWriter) Write 

func (bw *BufPipeWriter) Write(p []byte) (n int, err error)

Write implements the io.Writer interface: it writes data to the underlying storage. After filling the current block, it enqueues the block for the read side, and tries to continue writing to the next available block, which is provided by the read side.

Returning n (number of bytes written) means that the data is written to the underlying storage, read side reads the data from the storage asynchronously. To flush the last block to the storage, call Close() otherwise the last block will not be seen by the read side.

type ErrOrEofReader 

type ErrOrEofReader struct {
	Err error
}

ErrOrEofReader is an io.Reader implementation that always returns an error set in Err field or io.EOF if Err is nil.

func (*ErrOrEofReader) Read 

func (r *ErrOrEofReader) Read(p []byte) (int, error)

Read implements the io.Reader interface. It returns the error set in Err field or io.EOF if Err is nil.

type LruReaderAt added in v0.2.0 

type LruReaderAt[T io.ReaderAt] struct {
	// contains filtered or unexported fields
}

LruReaderAt wraps an io.ReaderAt and caches its data in an LRU cache. It is designed for reading random offsets efficiently, making it suitable for scenarios where repeated reads from non-contiguous regions occur. Use NewLruReaderAt to create an instance for your underlying reader. Underlying reader should not be modified while the LruReaderAt is in use to avoid data inconsistency.

It is safe for concurrent use, but method calls are synchronized with a mutex.

See ReadAt method for the read semantics.

Example (Simple) 
package main

import (
	"fmt"

	"github.com/ozanh/xio"
)

func main() {
	// Simple example of using LruReaderAt with xio.StorageBuffer.
	// Any io.ReaderAt can be used as the underlying reader.

	const helloWorld = "Hello, World!"

	rw := xio.NewStorageBuffer(make([]byte, 1024), false)
	n, err := rw.WriteAt([]byte(helloWorld), 0)
	if err != nil {
		panic(err)
	}
	if n != len(helloWorld) {
		panic("write failed")
	}

	const blockSize = 512
	const cacheSize = 10

	lruReader, err := xio.NewLruReaderAt(rw, blockSize, cacheSize)
	if err != nil {
		panic(err)
	}

	buf := make([]byte, len(helloWorld))
	n, err = lruReader.ReadAt(buf, 0)
	if err != nil {
		panic(err)
	}
	if n != len(helloWorld) {
		panic("read failed")
	}

	fmt.Printf("%s\n", buf)

}

Output:
Hello, World!

func NewLruReaderAt added in v0.2.0 

func NewLruReaderAt[T io.ReaderAt](reader T, blockSize, cacheSize int) (*LruReaderAt[T], error)

NewLruReaderAt creates a new CachingReaderAt with the given reader, blockSize, and cacheSize. blockSize and cacheSize should be a power of 2 for better performance. blockSize and cacheSize must be greater than 0. blockSize and cacheSize must be less than MaxUint32.

func (*LruReaderAt[T]) Metrics added in v0.2.0 

func (lra *LruReaderAt[T]) Metrics() LruReaderAtMetrics

Metrics returns the current metrics of the LruReaderAt. Purge resets these metrics.

func (*LruReaderAt[T]) Purge added in v0.2.0 

func (lra *LruReaderAt[T]) Purge()

Purge purges the underlying lru cache, and resets the metrics.

func (*LruReaderAt[T]) ReadAt added in v0.2.0 

func (lra *LruReaderAt[T]) ReadAt(p []byte, offset int64) (n int, err error)

ReadAt implements the io.ReaderAt and reads len(p) bytes into p starting at offset, using the cache where possible. If a cached block doesn’t fully satisfy the request, it reads the remainder from the underlying reader.

If number of read bytes is equal to the len(p), it always returns nil error if EOF was reached.

If the underlying reader does not implement the io.ReaderAt interface correctly and error is nil when n < len(p) , it does not cache the read bytes that are less than the block size, and returns without reading the remaining bytes.

func (*LruReaderAt[T]) Reset added in v0.3.2 

func (lra *LruReaderAt[T]) Reset(reader T)

Reset resets the LruReaderAt with a new reader and purges the cache for reuse.

type LruReaderAtMetrics added in v0.2.0 

type LruReaderAtMetrics struct {
	CacheInserts    uint64
	CacheCollisions uint64
	CacheEvictions  uint64
	CacheRemovals   uint64
	CacheHits       uint64
	CacheMisses     uint64
	CacheHitBytes   uint64
	PoolAllocs      uint64
	CachedCount     uint64
}

LruReaderAtMetrics contains the metrics of an LruReaderAt.

type ReadCounter 

type ReadCounter interface {
	io.Reader
	Count() uint64
}

ReadCounter is an interface that extends io.Reader with a Count method. The Count method returns the total number of bytes read from the reader.

type ReadDataAtomicCounter 

type ReadDataAtomicCounter[T io.Reader] struct {
	C atomic.Uint64
	R T
}

ReadDataAtomicCounter is a wrapper for an io.Reader that counts the number of bytes read. The Count method returns the total number of bytes read from the reader. Call Count to get total number of bytes read. Note: Count() is goroutine-safe.

func NewReadDataAtomicCounter 

func NewReadDataAtomicCounter[T io.Reader](r T) *ReadDataAtomicCounter[T]

func (*ReadDataAtomicCounter[T]) Count 

func (r *ReadDataAtomicCounter[T]) Count() uint64

Count returns the total number of bytes read from the reader. It is goroutine-safe.

func (*ReadDataAtomicCounter[T]) Read 

func (r *ReadDataAtomicCounter[T]) Read(p []byte) (int, error)

Read implements the io.Reader interface.

type ReadDataCounter 

type ReadDataCounter[T io.Reader] struct {
	C uint64
	R T
}

ReadDataCounter is a wrapper for an io.Reader that counts the number of bytes read. The Count method returns the total number of bytes read from the reader. Call Count to get total number of bytes read. Note: Count() is not goroutine-safe. Use ReadDataAtomicCounter[T] for goroutine-safe counting.

func NewReadDataCounter 

func NewReadDataCounter[T io.Reader](r T) *ReadDataCounter[T]

NewReadDataCounter returns a new ReadDataCounter that wraps the given reader.

func (*ReadDataCounter[T]) Count 

func (r *ReadDataCounter[T]) Count() uint64

Count returns the total number of bytes read from the reader. It is not goroutine-safe.

func (*ReadDataCounter[T]) Read 

func (r *ReadDataCounter[T]) Read(p []byte) (int, error)

Read implements the io.Reader interface.

type ReadersDataCmpError added in v0.3.0 

type ReadersDataCmpError struct {
	Err        error
	ErrLeft    error
	ErrRight   error
	BytesLeft  []byte
	BytesRight []byte
	Offset     int64
}

ReadersDataCmpError is an error returned by CmpReadersData when the data read from the readers is not equal.

func (*ReadersDataCmpError) Error added in v0.3.0 

func (e *ReadersDataCmpError) Error() string

func (*ReadersDataCmpError) Unwrap added in v0.3.0 

func (e *ReadersDataCmpError) Unwrap() error

type Storage 

type Storage interface {
	io.ReaderAt
	io.WriterAt
}

Storage is the interface that wraps the basic io.ReaderAt and io.WriterAt interfaces.

type StorageBuffer 

type StorageBuffer struct {
	// contains filtered or unexported fields
}

StorageBuffer implements the Storage interface using a single byte slice. It is for testing, debug or small data storage. StorageBuffer methods are goroutine-safe.

func NewStorageBuffer 

func NewStorageBuffer(buf []byte, autoGrow bool) *StorageBuffer

NewStorageBuffer creates a new StorageBuffer with the given byte slice and auto grow flag. StorageBuffer is suitable for small data storage and testing, for large data storage, use BlockStorageBuffer.

func (*StorageBuffer) AutoGrow 

func (s *StorageBuffer) AutoGrow() bool

AutoGrow reports whether the storage buffer is auto growing, which is set when creating the storage buffer.

func (*StorageBuffer) Bytes 

func (s *StorageBuffer) Bytes() []byte

Bytes returns the underlying byte slice.

func (*StorageBuffer) Len 

func (s *StorageBuffer) Len() int

Len returns the length of the underlying byte slice.

func (*StorageBuffer) ReadAt 

func (s *StorageBuffer) ReadAt(p []byte, off int64) (int, error)

ReadAt reads len(p) bytes from the storage starting at byte offset off.

func (*StorageBuffer) WriteAt 

func (s *StorageBuffer) WriteAt(p []byte, off int64) (int, error)

WriteAt writes len(p) bytes to the storage starting at byte offset off.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.