.Get() on pool not following the pool capacity

[praveen001]

To understand how the pools work I created a pool with capacity of 3.

        pool, err := faktory.NewPool(3)
 	if err != nil {
		log.Fatalln("Unable to connect to Faktory", err.Error())
	}
	log.Println("Pool created, length = ", pool.Len()) // length = 0

Then I called .Get() 10 times, expecting it to fail from the fourth call to .Get(). But all 10 calls succeeded.

        cli := []*faktory.Client{}
	for i := 0; i < 10; i++ {
		c, err := pool.Get()
		if err != nil {
			log.Fatalln("Unable to get client from pool", err.Error()) // never reached
		}
		cli = append(cli, c)
		fmt.Println("Got client ", i+1)
		fmt.Println("Length of pool", pool.Len())
	}

I was even able to push jobs using all 10 *faktory.Client.

        for i := 0; i < 10; i++ {
		job := faktory.NewJob("lazy", i)
		cli[i].Push(job) // succeeded, job was seen from web ui
	}

Finally I tried to return all of them, but only 5 clients were returns.

	for i := 0; i < 10; i++ {
		c := cli[i]
		pool.Put(c)
		fmt.Println("Got client ", i+1)
		fmt.Println("Length of pool", pool.Len()) // didn't go beyond 5.
	}

Is this an expected behavior? If yes, how to control the total number of connections?

[mperham]

Please give me the actual script to reproduce the behavior.

[praveen001]

I'm running faktory server with docker

docker run --rm -it -p 127.0.0.1:7419:7419 -p 127.0.0.1:7420:7420 contribsys/faktory:latest

Following script should help you reproduce the issue.

package main

import (
	"log"

	faktory "github.com/contribsys/faktory/client"
	worker "github.com/contribsys/faktory_worker_go"
)

// LazyWorker ..
func LazyWorker(ctx worker.Context, args ...interface{}) error {
	log.Println("Executing Job ", ctx.Jid(), args)
	return nil
}

func produce() {
	log.Println("Connecting to faktory")

	pool, err := faktory.NewPool(5)
	if err != nil {
		log.Fatalln("Unable to create to Faktory pool", err.Error())
	}

	// Array to hold 10 clients
	cli := []*faktory.Client{}

	// Get 10 clients continuously
	for i := 0; i < 10; i++ {
		c, err := pool.Get()
		if err != nil { // Expecting to hit this case on 6th iteration or it should wait for a clients to be returned to the pool?
			log.Fatalln("Unable to get client from pool", i, err.Error())
		}
		log.Println("Got client ", i)
		log.Println("Pool length ", pool.Len())
		cli = append(cli, c)
	}

	// Schedule 1 job with all 10 clients
	for i := 0; i < 10; i++ {
		job := faktory.NewJob("lazy", i)
		if err := cli[i].Push(job); err != nil {
			log.Fatalln("Unable to push job", i, err.Error())
		}
		log.Println("Scheduled job ", i)
	}

	// Return all 10 jobs to the pool
	for i := 0; i < 10; i++ {
		pool.Put(cli[i])
		log.Println("Returned client ", i)
		log.Println("Pool length ", pool.Len())
	}
}

func consume() {
	mgr := worker.NewManager()

	mgr.Concurrency = 1

	mgr.Register("lazy", LazyWorker)

	mgr.Run()
}

func main() {
        produce()
	consume()
}

OUTPUT:

2020/03/25 12:58:08 Connecting to faktory
2020/03/25 12:58:08 Got client  0
2020/03/25 12:58:08 Pool length  0
2020/03/25 12:58:08 Got client  1
2020/03/25 12:58:08 Pool length  0
2020/03/25 12:58:08 Got client  2
2020/03/25 12:58:08 Pool length  0
2020/03/25 12:58:08 Got client  3
2020/03/25 12:58:08 Pool length  0
2020/03/25 12:58:08 Got client  4
2020/03/25 12:58:08 Pool length  0
2020/03/25 12:58:08 Got client  5
2020/03/25 12:58:08 Pool length  0
2020/03/25 12:58:08 Got client  6
2020/03/25 12:58:08 Pool length  0
2020/03/25 12:58:08 Got client  7
2020/03/25 12:58:08 Pool length  0
2020/03/25 12:58:08 Got client  8
2020/03/25 12:58:08 Pool length  0
2020/03/25 12:58:08 Got client  9
2020/03/25 12:58:08 Pool length  0
2020/03/25 12:58:08 Scheduled job  0
2020/03/25 12:58:08 Scheduled job  1
2020/03/25 12:58:08 Scheduled job  2
2020/03/25 12:58:08 Scheduled job  3
2020/03/25 12:58:08 Scheduled job  4
2020/03/25 12:58:08 Scheduled job  5
2020/03/25 12:58:08 Scheduled job  6
2020/03/25 12:58:08 Scheduled job  7
2020/03/25 12:58:08 Scheduled job  8
2020/03/25 12:58:08 Scheduled job  9
2020/03/25 12:58:08 Returned client  0
2020/03/25 12:58:08 Pool length  1
2020/03/25 12:58:08 Returned client  1
2020/03/25 12:58:08 Pool length  2
2020/03/25 12:58:08 Returned client  2
2020/03/25 12:58:08 Pool length  3
2020/03/25 12:58:08 Returned client  3
2020/03/25 12:58:08 Pool length  4
2020/03/25 12:58:08 Returned client  4
2020/03/25 12:58:08 Pool length  5
2020/03/25 12:58:08 Returned client  5
2020/03/25 12:58:08 Pool length  5
2020/03/25 12:58:08 Returned client  6
2020/03/25 12:58:08 Pool length  5
2020/03/25 12:58:08 Returned client  7
2020/03/25 12:58:08 Pool length  5
2020/03/25 12:58:08 Returned client  8
2020/03/25 12:58:08 Pool length  5
2020/03/25 12:58:08 Returned client  9
2020/03/25 12:58:08 Pool length  5
2020/03/25 12:58:08 Executing Job  iYCF7xJOTtmV1336 [0]
2020/03/25 12:58:08 Executing Job  ztHOx-fnVNtp1gcw [1]
2020/03/25 12:58:08 Executing Job  9E69YN5HRHOnwse0 [2]
2020/03/25 12:58:08 Executing Job  KwYsQgxA00MdJVwh [3]
2020/03/25 12:58:08 Executing Job  V3BGLdWmJYov3Emk [4]
2020/03/25 12:58:08 Executing Job  SIXiLYs3Lm0PNPGE [5]
2020/03/25 12:58:08 Executing Job  ja99cafBH-qFcabT [6]
2020/03/25 12:58:08 Executing Job  2vyJdcSp9uwzM4Ac [7]
2020/03/25 12:58:08 Executing Job  AybXRXzKHm4TXn_t [8]
2020/03/25 12:58:08 Executing Job  tUWaiVNNocw3Xp7K [9]

Please let me know, If I'm doing something wrong.

[mperham]

Nevermind, I'm working on it.

[mperham]

Yeah, looks like this pool impl (that I copied from fatih/pool) doesn't actually limit the number of connections, it only limits the number of connections held in the pool. Any extra will be created on demand, no matter the maximum pool capacity. Arguably a bug, arguably a feature.