Store variables per worker

[Azer0s]

I was wondering if there was any built-in way of storing data per worker (e.g. TCP connections) and retrieve them in a task.
Something like this:

// Create a pool with limited concurrency
pool := pond.NewPool(100)

// Submit 1000 tasks
for i := 0; i < 1000; i++ {
	i := i
	pool.Submit(func(wc pond.WorkerContext) {
                t, ok := wc.Get("worker")
                if !ok {
                        t = i
                        wc.Set("worker, i)
                }

		fmt.Printf("Running task #%d in worker #%d\n", i, t)
	})
}

I don't know a lot about pond but I think such a feature would be quite cool

[alitto]

Hi @Azer0s!

At the moment, this library doesn't provide any built-in mechanism to "store" state on a worker.

Under the hoods, each worker is simply a goroutine running the worker function (internal). Unlike other concurrency abstractions ( e.g. "threads" in Java), Golang goroutines are meant to be lightweight and the language itself doesn't provide a mechanism to attach custom variables/state to it. Moreover, there's native way to retrive an identifier of a particular goroutine.

Additionally, workers created by this library are killed immediately if they become idle, so any state associated with them would be removed too.

I'm curious about your real use case though, what kind of data would you store in the worker's state?

[Azer0s]

Hi!

Thanks for the quick response and explanation 😊

Let me explain the motivation a bit more clearly.

In some environments - especially in more "legacy-heavy" contexts like government or finance - we sometimes need to interact with systems that are very slow to establish a connection with. Think old-school mainframes or enterprise backends that use something like IBM MQ or even TN3270 emulation.

In those cases, connection pooling is essential. But it's not just about reusing connections arbitrarily - we often want to pin a connection to a specific worker to avoid synchronization overhead or meet certain affinity requirements (e.g., session state bound to a specific connection).

My idea was to provide a mechanism to store some state per worker (or a persistent state that can only ever be assigned to a single worker at a time) - like an open TCP connection - so when a task is assigned to that worker again, it can reuse that state instead of recreating it.

[Azer0s]

So something like

wcPool := pond.NewContextPool[net.Conn](100, "conn")
for i := 0; i < 100; i++ {
	wcPool.Add(establishConn())
}

// Create a pool with limited concurrency
pool := pond.NewPool(100, pond.WithContextPool(wcPool))

// Submit 1000 tasks
for i := 0; i < 1000; i++ {
	i := i
	pool.Submit(func(wc pond.WorkerContext) {
                // will block until a conn is available
                // conn will be unlocked after the function so it can be reused for another "Submit"
                c := wc.Get[net.Conn]("conn")

                doConn(c)
	})
}

[alitto]

I think your use case is better suited with sync.Pool, which is part of the standard sync package: https://pkg.go.dev/sync#Pool.
From the struct's description:

A Pool is a set of temporary objects that may be individually saved and retrieved.
An appropriate use of a Pool is to manage a group of temporary items silently shared among and potentially reused by concurrent independent clients of a package. Pool provides a way to amortize allocation overhead across many clients.

Here's how you could use it to implement a pool of "expensive" connections that can be reused by at most 100 concurrent clients. Notice that the max concurrency of 100 in this case is guaranteed by pond's worker pool, so you don't need to specify the max number of connections in sync.Pool (there cannot be more than 100 goroutines requesting connections at the same time).

import (
    "sync"
    "net"
    "github.com/alitto/pond/v2"
)

// Create a connection pool using sync.Pool
connPool := sync.Pool{
        New: func() any {
            return establishConn()
        },
}

// Create a pool with limited concurrency
pool := pond.NewPool(100)

// Submit 1000 tasks
for i := 0; i < 1000; i++ {
    pool.Submit(func(wc pond.WorkerContext) {
        conn := connPool.Get().(net.Conn)
        defer connPool.Put(conn)

	doConn(conn)
    })
}

Since connections are established when retrieved (sync.Pool executes the New func at that moment), you don't need to create them in advance. But if you need to do that for some reason, you could do it like this:

// Create a connection pool using sync.Pool
connPool := sync.Pool{
        New: func() any {
            return establishConn()
        },
}

// Force pre-creation of 100 connections
conns := make(net.Conn[], 0)
for i := 0; i < 100; i++ {
    conn := connPool.Get().(net.Conn)
    conns := append(conns, conn)
}
for conn := range conns {
    connPool.Put(conn)
}

The only caveat here is sync.Pool may deallocate items if GC needs to free memory, but this could prove very useful if running in a memory constrained environment.

Please let me know your thoughts on this.
Cheers