Forem: Bartłomiej Święcki

Quick way to combine io.Reader and io.Closer

Bartłomiej Święcki — Sun, 22 Jan 2023 00:00:00 +0000

There are many interesting tools in Golang's standard library to wrap io.Reader instance such as io.LimitedReader or cipher.StreamReader. But when wrapping a io.ReadCloser instance, the Close method is hidden.

Here's a quick code snippet to combine wrapped io.Reader and the original io.Closer through an inline struct to rebuild the io.Closer interface.

Code

var rc io.ReadCloser = struct {
    io.Reader
    io.Closer
}{
    Reader: r,
    Closer: c,
}

What it is about?

The io.Reader interface in Golang is a very powerful abstraction when streaming data. It can be used to read files, http responses, even raw byte arrays using a generic code:

func sumAllBytes(r io.Reader) (uint64, error) {
    var buff [512]byte
    var sum uint64

    for {
        n, err := r.Read(buff[:])
        for i := 0; i < n; i++ {
            sum += uint64(buff[i])
        }

        if errors.Is(err, io.EOF) {
            return sum, nil
        }
        if err != nil {
            return sum, err
        }
    }
}

That method can be used to process files, http responses, even memory buffers:

func main() {

    // Process a buffer
    buff := bytes.NewReader([]byte{0x01, 0x02, 0x03, 0x04})

    sum, err := sumAllBytes(buff)
    if err != nil {
        log.Fatal(err)
    }
    log.Println("Sum from buffer:", sum)

    // Process a http response

    resp, err := http.Get("https://www.google.com")
    if err != nil {
        log.Fatal(err)
    }
    defer resp.Body.Close()

    sum, err = sumAllBytes(resp.Body)
    if err != nil {
        log.Fatal(err)
    }
    log.Println("Sum from http:", sum)

    // Process a file

    fl, err := os.Open("/some/file/path")
    if err != nil {
        log.Fatal(err)
    }
    defer fl.Close()

    sum, err = sumAllBytes(fl)
    if err != nil {
        log.Fatal(err)
    }
    log.Println("Sum from file:", sum)
}

Cleaning up

In many cases (like the http and file instances above), the reader has to be explicitly closed to avoid resource leaks. For that reason, many resources are using the io.ReadCloser interface and cleaning up can easily be achieved with a defer rc.Close() statement. But there are some cases where the close method is not called at the same function but somewhere at the caller site. At that construct, the caller is responsible for cleanup:

func getDataStream(name string) (io.ReadCloser, error) {
    switch name {
    case "file":
        return os.Open("/some/file")
    case "http":
        resp, err := http.Get("https://www.google.com/")
        if err != nil {
            return nil, err
        }
        return resp.Body, nil
    case "buffer":
        return io.NopCloser(
            bytes.NewReader([]byte{0x01, 0x02, 0x03, 0x04}),
        ), nil
    default:
        return nil, errors.New("Invalid data stream name")
    }
}

func printSum(streamName string) {
    stream, err := getDataStream(streamName)
    if err != nil {
        log.Fatal(err)
    }
    defer stream.Close()

    sum, err := sumAllBytes(stream)
    if err != nil {
        log.Fatal(err)
    }

    log.Printf("Sum of bytes in stream '%s' is '%d\n", streamName, sum)
}

So far so good, nothing to worry about. But let's extend this example with some stream wrapping:

func getDataStream(name string) (io.ReadCloser, error) {
    switch name {
    case name == "file":
        return os.Open("/some/file")
    case name == "http":
        resp, err := http.Get("https://www.google.com/")
        if err != nil {
            return nil, err
        }
        return resp.Body, nil
    case name == "buffer":
        return io.NopCloser(
            bytes.NewReader([]byte{0x01, 0x02, 0x03, 0x04}),
        ), nil

    // v---  Create a truncated stream by applying limit over the base one ---v
    case strings.HasPrefix(name, "limit:"):
        r, err := getDataStream(name[6:])
        if err != nil {
            return nil, err
        }
        return io.LimitReader(r, 3)

    default:
        return nil, errors.New("Invalid data stream name")
    }
}

Unfortunately this code does not compile and ends up with this error:

cannot use io.LimitReader(r, 100) (value of type io.Reader) as type io.ReadCloser in return statement:
    io.Reader does not implement io.ReadCloser (missing Close method)

Wrapping the io.ReadCloser with io.LimitedReader does hide the io.Closer functionality of the original instance. And it turns out that there are many places in golang standard lib where such wrapping takes place.

Inline struct to the rescue

There's an easy trick to bring back the Close method from the original reader back to the wrapped one:

func getDataStream(name string) (io.ReadCloser, error) {
    switch name {
    case name == "file":
        return os.Open("/some/file")
    case name == "http":
        resp, err := http.Get("https://www.google.com/")
        if err != nil {
            return nil, err
        }
        return resp.Body, nil
    case name == "buffer":
        return io.NopCloser(
            bytes.NewReader([]byte{0x01, 0x02, 0x03, 0x04}),
        ), nil

    // v---  Create a truncated stream by applying limit over the base one ---v
    case strings.HasPrefix(name, "limit:"):
        r, err := getDataStream(name[6:])
        if err != nil {
            return nil, err
        }

        limitReader := io.LimitReader(r, 3)

        return struct {
            io.Reader
            io.Closer
        }{
            Reader: limitReader, // Read method will come from the wrapped reader
            Closer: r,           // Close method will come from the original reader
        }, nil

    default:
        return nil, errors.New("Invalid data stream name")
    }
}

How does it work?

The inline struct contains two embedded fields, one for the reader and the other for the closer.

Since those fields are anonymous, the struct itself inherits methods from those fields as if those were declared on the struct. By doing so, whenever the compiler tries to cast the struct to some interface, it can promote those methods to fulfil the requirements of the interface.

In the code above we return an instance of io.ReadCloser interface that requires both Read and Close methods - and those are borrowed from embedded fields respectively.

Interestingly, if we would use whole io.ReadCloser as the second embedded field instead of io.Reader, the compiler (go 1.19 as of writing) throws an error which is caused by ambiguity between promoted field members (the Read method is not promoted due to ambiguity):

cannot use struct{io.Reader; io.ReadCloser}{…} (value of type struct{io.Reader; io.ReadCloser}) as type io.ReadCloser in return statement:
    struct{io.Reader; io.ReadCloser} does not implement io.ReadCloser (missing Read method)

Throttling Pool

Bartłomiej Święcki — Tue, 22 Nov 2022 00:00:00 +0000

In my last post I presented few techniques useful for limiting simultaneous operations in Go. Such throttlers are indeed a very neat tools to keep our resources under control. However we can be extended that idea to even better tool for resource control when we mix throttlers with object pools.

Object pools in go

Even though golang has a very efficient memory management system, there are some cases where we would like to manage the memory by ourselves. There may be many reasons for that. One may want to ensure that the amount of allocated memory does not cross some predefined limit. It may also be the case that frequent allocations and deallocations of memory start causing issues due to a huge CPU cost. Those factors are especially important in more complex systems with tight memory budget such as databases. And it turns out that I had a chance to work with one such database called immudb for more than a year now.

In immudb there are some larger scratchpad objects used during the transaction commit that are preallocated when the database is loaded. Once a writer starts preparing a transaction, such scratchpad object is taken from the pool to be returned back to it once the commit finishes.

Sounds familiar? Looks almost identical to the technique used for throttling the number of simultaneous operations with tokens that I presented in the previous post. But instead of fetching a token from the pool that is of no use, we can fetch a useful object. This mechanism is used by immudb to keep the number of simultaneous transactions under control. Without such limit, it would be easy to force the database to use huge amount of resources (which some may refer to as a DoS attack if for example the database runs out of memory).

The implementation in immudb is based on a mutex whereas in this blog post I'll focus on a channel-based implementation that will be similar to what we've seen with token-based throttlers.

Pool-throttler

Let's do some exercise now and create such a throttling pool.

I decided to use generics in the implementation. Instead of creating a pool that works on the interface{} type (or some other concrete one) we can leave the selection of that underlying type to the user of the pool. That way we can avoid a lot of conversions and type assertions ruling out unnecessary casting cost.

As with tokens, the core of pool-based throttler will again be a channel:

type poolLimiter[T any] struct {
    ch chan T
}

In case of token-based throttlers the initialization of such channel was simple because the token type was empty and did not contain any information. In case of a pool, objects in the pool must be initialized correctly. For that reason I decided to add a dedicated initialization functor when creating objects in the pool:

func NewPoolLimiter[T any](poolSize int, gen func() T) PoolLimiter[T] {

    ret := &poolLimiter[T]{
        ch: make(chan T, poolSize),
    }

    // Preallocate the pool of objects
    for i := 0; i < poolSize; i++ {
        ret.ch <- gen()
    }

    return ret
}

And now the essence of the pool - acquiring and releasing resources. Let's write few implementations for different use cases, starting with the simplest one.

Blocking acquire

The easiest acquire method to write is the blocking one that will wait until there is at least one object in the pool:

func (p *poolLimiter[T]) Acquire() T {
    return <-p.ch
}

Nothing really fancy here - the channel gives us exactly the functionality that we need. But you may also have noticed that contrary to the previous token-based implementations I decided not to return the release function from that method.

Instead there will be a dedicated Release function in the pool itself:

func (p *poolLimiter[T]) Release(item T) {
    p.ch <- item
}

Why such change of the interface? I did some benchmarks and this way the implementation is almost twice as fast (41 ns/op vs 72 ns/op). The overhead of additional function value is pretty significant here. For that reason I also changed the interface of the original token-based throttler (that code even simplified which is usually a good sign).

As a side note, the code written so far looks too trivial, isn't it? Well, it just proves that golang's built-in primitives are really well designed.

Non-blocking acquire

Let's move to something a bit more complex now - acquire that will return an error if the pool is empty:

func (p *poolLimiter[T]) AcquireNoWait() (T, error) {
    var item T
    select {
    case item = <-p.ch:
        return item, nil
    default:
        return item, ErrResourceExhausted
    }
}

Nothing really fancy here - the built-in select is all we need.

The returned item is created and initialized at the beginning of this method which may be a bit surprising. This is needed if the pool is empty. In such case we still have to return some instance of the T type along with an error. Creating a default value initialized at the beginning does the trick.

Context-aware acquire

The most complex (but still fitting in just a few lines of code) is the context-aware acquire:

func (p *poolLimiter[T]) AcquireCtx(ctx context.Context) (T, error) {
    var item T

    if ctx.Err() == nil {
        select {
        case item = <-p.ch:
            return item, nil
        case <-ctx.Done():
            break
        }
    }

    return item, ctx.Err()
}

The code is almost identical to the non-blocking acquire. If the pool is empty and we have to wait for some freed up resources, catching context cancellation with the select built-in allows the function to give up and return an error. Additional if ctx.Err() == nil ensures that we won't even try to fetch the resource if the context has already been cancelled.

Exercise - token from pool

We can try to compare the newly written pool against the older token limiter. For that reason I've created a simple wrapper struct encapsulating the pool but exposing the token limiter interface instead:

type poolLimiterToTokenLimiter struct {
    l PoolLimiter[struct{}]
}

func NewTokenLimiterFromPoolLimiter(maxTokens int) CancellableTokenLimiter {
    return &poolLimiterToTokenLimiter{
        l: NewPoolLimiter(maxTokens, func() struct{} { return struct{}{} }),
    }
}

func (l *poolLimiterToTokenLimiter) Acquire() {
    l.l.Acquire()
}

func (l *poolLimiterToTokenLimiter) AcquireNoWait() error {
    _, err := l.l.AcquireNoWait()
    return err
}

func (l *poolLimiterToTokenLimiter) AcquireCtx(ctx context.Context) error {
    _, err := l.l.AcquireCtx(ctx)
    return err
}

func (l *poolLimiterToTokenLimiter) Release() {
    l.l.Release(struct{}{})
}

Such implementation was roughly 30% slower than the dedicated token limiter implementation (48.54 ns/op vs 36.90 ns/op). My guess is that this is mostly caused by indirect calls and additional returned value. I think that the gap will become smaller in future versions of the go compiler that will be able to inline functions and detect dead code much better.

The topic of throttling is still not fully explored here so stay tuned for more updates.

And as before, you can find the full source code with all test and benchmarks in Github repository.

Throttling in Go

Bartłomiej Święcki — Fri, 07 Oct 2022 00:00:00 +0000

Go language makes a good job at making hard things simple. It compiles the code so that it runs fast, it comes with garbage collector so we don't have to deal with memory management, it has lightweight goroutines and we don't have to think a lot about concurrency limits. And in majority of cases all those language constructs combined with sane defaults will be good enough. Just like writing a web server. Its almost never the case that we have to think about how many parallel requests we can handle.

But sometimes we need to have more control over our resources.

Exploring Restic

Few years ago, while I was still a newbie in the go world, I was exploring internals of Restic - a nice backup tool written in go. While doing so I learned a lot, but one thing caught my attention. There was an upper limit of simultaneous upload to a remote storage such as an s3 server.

When I first saw Restic in action I thought that this must have been done using some upload queue, maybe some upload scheduler or some other complicated machinery. But the implementation in Restic turned out to be much simpler. Each upload is realized in a separate goroutine and there are many of them spawned in parallel, much more than what is really necessary to saturate upload speed. But to avoid unbounded number of parallel uploads, there's a throttling mechanism based on upload tokens. Those tokens are acquired before sending data to the remote storage and released once the upload is done. The number of upload tokens is limited constraining the number of parallel uploads.

Since I was a golang newbie back then, I didn't really understand the code when I first looked at it. It was using channels so I thought that it is related to some communication between goroutines. I only realized that this is a very neat throttling mechanism a bit later.

Channels as token machines

How is it done? A dedicated go channel is used as a token disposal machine. To do some throttled operation, one takes the token from that channel (token := <- channel) and gives it back once the operation is finished (channel <- token). If there are no tokens left, the channel will naturally block the goroutine until someone else returns previously acquired token back to the pool.

All of that can be done in just few lines of go code.

In this post I'd like to play a bit with this idea and show you how easy such throttlers can be implemented.

The basic implementation

Let's start with some worker code. As we've discussed before the worker should first acquire the token and give it back to the pool once the task is finished:

func worker() {
    token := <- tokens                  // Grab the token, can block
    defer func() { tokens <- token }()  // Return the token

    // do the hard work...
}

Before we can use the tokens channel it must be initialized by filling it up with tokens:

const tokensCount = 30
var tokens = make(chan struct{}, tokensCount)

func initTokens() {
    for i:=0; i<tokensCount; i++ {
        tokens <- struct{}{}
    }
}

We can also extract the throttler to a separate easy to use object:

type Throttler struct {
    ch chan struct{}
}

func NewThrottler(max int) *Throttler {
    ch := make(chan struct{}, max)

    for i := 0; i < max; i++ {
        ch <- struct{}{}
    }

    return &Throttler{ch: ch}
}

func (c *Throttler) Acquire() func() {
    token := <-c.ch
    return func() { c.ch <- token }
}

and use it in our worker:

const tokensCount = 30
var tokens = NewThrottler(tokensCount)

func worker() {
    defer tokens.Acquire()()    // Throttle

    // do the hard work...
}

You may notice that the Acquire method returns a func() object instead of the token. That function will release the token back to the pool when executed. That way the acquire and the release of the token are tightly coupled together and we can both acquired the token and schedule its release upon function exit in a nice one-liner.

Reversed implementation

The implementation above can be interpreted as the channel being a set of idle tokens. Worker takes the token from that set and gives it back once done with the job.

But we can also reverse the idea and use channel as a holder of tokens for busy workers. Before doing the task, worker puts its token into the channel and takes it back once the work is finished. Since the channel has the maximum capacity, once it is full, more busy tokens will have to wait until someone takes one's token from the channel:

type Throttler struct {
    ch chan struct{}
}

func NewThrottler(max int) *Throttler {
    ch := make(chan struct{}, max)
    return &Throttler{ch: ch}
}

func (c *Throttler) Acquire() func() {
    c.ch <- struct{}{}
    return func() { <-c.ch }
}

The biggest difference here is that the channel doesn't have to be populated with tokens during initialization but otherwise there's not much of a difference between those two implementations.

Old-style implementation

In previous examples the core of the throttler was a go channel. But the same can be implemented using old-style synchronization primitives:

type Throttler struct {
    cnt int
    m   sync.Mutex
    c   sync.Cond
}

func NewThrottler(max int) *Throttler {
    ret := &Throttler{
        cnt: max,
    }
    ret.c.L = &ret.m
    return ret
}

func (c *Throttler) Acquire() func() {
    c.m.Lock()
    defer c.m.Unlock()

    for c.cnt <= 0 {
        c.c.Wait()
    }
    c.cnt--

    return func() {
        c.m.Lock()
        defer c.m.Unlock()

        c.cnt++
        c.c.Signal()
    }
}

Much more complex implementation compared to the previous one, right?

When I first wrote it I assumed that there will be a huge performance boost because we replaced a heavy channel object with a lightweight low-level synchronization mechanisms.

But after I run some benchmarks it turned out that this implementation is only between 12% to 19% faster than our original one. This does not sound like a huge difference to me. Maybe in some very specific CPU-sensitive use cases it could be beneficial but I believe that if each spent CPU cycle is making the difference then languages such as c or c++ (or even assembly) would be a better fit there.

Putting performance difference aside, there's one more huge difference I'd like to point out.

Cancellable waits

Our channel-based throttlers have one huge advantage over the one based on mutex. They can easily be extended to also support cancellation through context.

Here's an example of a cancellable Acquire:

func (c *Throttler) Acquire(ctx context.Context) (func(), error) {
    if ctx.Err() == nil {
        select {
        case <-c.ch:
            return func() { c.ch <- struct{}{} }, nil
        case <-ctx.Done():
            break
        }
    }
    return nil, ctx.Err()
}

If the context is cancelled for any reason, the token acquire will now simply return an error:

func worker(ctx context.Context) error {
    releaseToken, err := tokens.AcquireCtx(ctx)
    if err != nil {
        return err
    }
    defer releaseToken()

    // do the hard work...
}

Why we would need such context-aware throttler? Let's take a simple http server as an example. A http request can be cancelled at any time as a result of the client dropping the connection or server shutdown. Waiting for the token in case the request is already cancelled is most likely pointless and the request handler should return immediately.

Low-level does not always mean more powerful

Such cancellable acquire can not be easily implemented with low-level mutexes and condition variables. Locking a mutex or waiting on a condition in go can not be interrupted. This means that the channel-based version is not only easier to write and understand but also handles much broader set of use-cases.

We could of course try experimenting with some complex implementation of a mutex-based throttler supporting cancellation. But at this point I don't believe that we would gain much more with it. Considering relatively small performance gains I'd rather stay with the channel-based implementation.

Future improvements

I hope you liked my introduction to throttling in go. Of course I only scratched the surface of this topic so stay tuned for future posts where I'll show few more tips and tricks.

See you on GitHub

I've gathered various implementations of throttlers in this github repository. It's a well tested (aiming at 100% test coverage) tiny go module that anybody can use. It also contains benchmarks to check the performance difference between various implementations.

Multi-CPU Github Actions with Go

Bartłomiej Święcki — Tue, 19 Jul 2022 00:00:00 +0000

When we think about CPU architecture usually there's one leader that comes to our mind - the famous x86-64 one. It is the main player on our desktops and on the server side. Even more recent generation of game consoles switched to that architecture from some more exotic ones.

But there are alternatives. Some are pretty well known such as the ARM one that took over the mobile market (and slowly enters the desktop world with Apple M1 laptops). But there are many more like RISC-V or SPARC, each with special properties and target group usually in the enterprise segment.

It is not a surprise that even newer languages such as Go support variety of CPU architectures. A quick check on my local Go compiler reveals that the for the Linux OS itself it already supports 12 CPU architectures:

$ go tool dist list | grep '^linux' | wc -l
12

Automatic builds and tests with Github Actions

I can't imagine high quality project without CI pipeline checking if the code works as expected. With variety of tools and ready-to-use online services it's a must. Like Github actions. Setting a Github project with Github Actions is really simple. But if the target application should work on various CPU architectures, there's not much physical hardware easily available that we could choose from for our test workflows.

Github currently offers only x86-64 machines (with Windows, Linux or MacOS) which means that it will be hard to check if subtle CPU differences don't affect the behavior of our code. Those could be as simple as some assumption about the size of the int type, hardcoded endianness or different memory alignment.

Qemu to the rescue

Qemu is a well-known and established software emulating various CPU architectures. And it turns out that it can easily be used to execute Linux binaries built for different CPU architectures.

Before we get into details, let me share a project where I demonstrate how it can be used: https://github.com/byo/multi-cpu-demo/actions. I created it to demonstrate application written in go that is compiled and tested on various CPU architectures. The essence is in this workflow file. Let's take a look at its most relevant sections.

Setting up qemu

It turns out that there's already working Github action for that. Installing and setting up qemu for emulation is as simple as that:

...
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      ...
      - uses: docker/setup-qemu-action@v2
      ...

That setup-qemu-action step itself does all the trickery to ensure the correct qemu version is installed and tightly integrated with the host (runner) operation system (more on that later).

Compiling go code for target platform

Go compiler can easily generate binaries for various OSes and CPU architectures no matter what the host OS and architecture the compiler runs on.

To switch to different CPU we only have to setup the GOARCH environment variables:

...
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      ...
      - run: go build -o hello .
        env:
          GOARCH: arm
      ...

Changing the target operating system can be done in a very similar way - by setting the GOOS environment variable. We won't be dealing with different OS-es though, only Linux this time 🐧😉.

Executing binaries with emulation

This is something that I was really surprised with. The qemu instance set up with docker/setup-qemu-action@v2 is actually installed in a very clever way. It does not simply install qemu binaries, actually I think it does not even copy a single binary to the host OS. Instead it is using binfmt_misc kernel mechanism to integrate qemu as an emulator needed to run certain binaries - in our case those are binary files for architecture other than the host system. If we try to run binary not meant for the host CPU, qemu takes over and does all the magic behind the scenes:

...
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      ...
      - run: ./hello
      ...

Running go tests on different CPU architecture

The tight system integration of qemu has one more advantage. It allows running go tests just as if we were running those on the host CPU:

...
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      ...
      - run: go test -v .
        env:
          GOARCH: arm
      ...

How does it work then? Does it mean that we're running the go toolchain itself in qemu?

Well, not really. When the go test command is executed, what it is actually doing is to compile a temporary binary from the test code and then executes it.

The go command itself is built for the host CPU which is x86-64 in this case. It first runs on the host CPU and starts the test compilation process and here the GOARCH environment variable jumps in. The compiled temporary binary is built for the GOARCH target CPU architecture (arm in our example above).

When the go compiler tries then to execute that binary, qemu jumps in (through the tight integration with the host OS) and thus the test itself runs on CPU emulated with qemu. For the go command this is totally transparent.

Trade-offs and gotchas

With the presented method we can run binaries on variety of different CPU architectures. But this does not use the OS emulation itself. The only thing we can change is the GOARCH variable and the GOOS must be left as linux. This can usually be worked around by using VM emulation or choosing different OS for the runner itself. There are cases like the Apple M1 though where I didn't find so far a reliable way to run tests without buying a dedicated hardware box.

Also keep in mind that CPU emulation is a very heavy process itself. Binaries executed through emulation will run much slower than on the real hardware. There could also be subtle differences between the real and the emulated CPU.

It's also possible that the emulation itself will contain bugs. I experienced that myself before and even posted a bogus bug report to the go compiler team only to figure out later that this was a bug in the qemu version that I was using.

Another fact worth mentioning here is that qemu does not emulate all CPU architectures supported by the go compiler (take a look at this commit from my example project). This means that some less popular configurations will still need either manual testing or dedicated hardware.