Connection Pooling

[hamiltop]

Poolboy

Poolboy works by providing exclusive access to a given resource in a "transaction".

Pros:

• Dynamic connection allocation is taken care of for you.

Cons:

• Connection affinity is required for cursors, which means a transaction may be required for the full length of the cursor. For long running change feeds, this will require one connection per changefeed.
• Pipelining is not utilized as each connection is checked out until a response is provided.

While my assumption was to use Poolboy, it seems that there are better approaches available. There are too many tradeoffs to be made for not a whole lot of benefit.

Hand rolled

An ideal connection pool will do the following:

1. Connect to N clients in a cluster.
2. Load balance queries between connections.
3. Route query to master replica for the target table.
4. If a connection dies, remove that connection from the rotation.
5. If a query fails because of a server connectivity issue, retry query on another host.

Initial design:

A supervisor tree will be over N connections and a coordinator.

Client processes send queries directly to the connection, just like we currently do. In order to know which connection to contact, the client can request a connection from the coordinator. When the client is finished with the query it informs the coordinator (in order for the coordinator to properly balance connections).

This would require zero changes to Exrethinkdb.Connection. The connection would be dumb to pooling. We could provide a Exrethinkdb.ConnectionPool module with run/next/close that wrap the logic for interacting with the coordinator. The end API for the user would be the same. We will also provide a use Exrethinkdb.ConnectionPool macro. Ideally you would replace use Exrethinkdb.Connection with use Exrethinkdb.ConnectionPool and your application would still work perfectly.

Routing queries to proper replicas can be done fairly well in the coordinator by looking at table("foo_table") |> config |> run. The coordinator can do this on a timer (once every minute is probably sufficient). It can then use this information to route the connection properly. In the event of stale_reads, the coordinator can load balance between non-master connections. We'll have to add table to the Query struct, but that's pretty straightforward.

Exrethinkdb.ConnectionPool.run will report back to the coordinator the success or failure of a query. The coordinator will also be monitoring the Connection processes. There will also be logic to retry on a different server if a failure occurs (in the event of stale_reads. otherwise, retrying won't be useful until a new master is selected).

I think this custom approach will be fairly simple. It will be designed so that failure leads to a worst case scenario of routing requests to the less optimal replica.

[jeregrine]

Their is an issue open for Ecto that solves this by "reserving" a worker
elixir-ecto/ecto#569 I'd be pretty strongly
against hand rolling because this kind of thing is full of pitfalls. We
could also ping @fishcakez and see if maybe his sbroker might be a stronger
use case here.

On Wed, May 6, 2015 at 11:15 PM, Peter Hamilton notifications@github.com
wrote:

Poolboy

Poolboy works by providing exclusive access to a given resource in a
"transaction".

Pros:

• Dynamic connection allocation is taken care of for you.

Cons:

• Connection affinity is required for cursors, which means a
  transaction may be required for the full length of the cursor. For long
  running change feeds, this will require one connection per changefeed.
• Pipelining is not utilized as each connection is checked out until a
  response is provided.

While my assumption was to use Poolboy, it seems that there are better
approaches available. There are too many tradeoffs to be made for not a
whole lot of benefit.
Hand rolled

An ideal connection pool will do the following:

1. Connect to N clients in a cluster.
2. Load balance queries between connections.
3. Route query to master replica for the target table.
4. If a connection dies, remove that connection from the rotation.
5. If a query fails because of a server connectivity issue, retry
   query on another host.

Initial design:

A supervisor tree will be over N connections and a coordinator.

Client processes send queries directly to the connection, just like we
currently do. In order to know which connection to contact, the client can
request a connection from the coordinator. When the client is finished with
the query it informs the coordinator (in order for the coordinator to
properly balance connections).

This would require zero changes to Exrethinkdb.Connection. The connection
would be dumb to pooling. We could provide a Exrethinkdb.ConnectionPool
module with run/next/close that wrap the logic for interacting with the
coordinator. The end API for the user would be the same. We will also
provide a use Exrethinkdb.ConnectionPool macro. Ideally you would replace use
Exrethinkdb.Connection with use Exrethinkdb.ConnectionPool and your
application would still work perfectly.

Routing queries to proper replicas can be done fairly well in the
coordinator by looking at table("foo_table") |> config |> run. The
coordinator can do this on a timer (once every minute is probably
sufficient). It can then use this information to route the connection
properly. In the event of stale_reads, the coordinator can load balance
between non-master connections. We'll have to add table to the Query
struct, but that's pretty straightforward.

Exrethinkdb.ConnectionPool.run will report back to the coordinator the
success or failure of a query. The coordinator will also be monitoring the
Connection processes. There will also be logic to retry on a different
server if a failure occurs (in the event of stale_reads. otherwise,
retrying won't be useful until a new master is selected).

I think this custom approach will be fairly simple. It will be designed so
that failure leads to a worst case scenario of routing requests to the less
optimal replica.

Reply to this email directly or view it on GitHub
#32.

[hamiltop]

I think table affinity is not going to be provided by existing solutions.
Table affinity will cut down on network bottlenecks on the server side.

The ecto solution solves a very different problem of nested transactions.
The solution they use (correct me if I'm wrong) is to permanently checkout
a connection for a long lived process to use. The alternatives proposed are
to provide the worker in the gen_call.

We are dealing with a very different set of constraints. The unit of
exclusivity for a postgres connection is a full query/response or even a
full transaction. For rethinkdb it's a query or a response. Since we can
interleave queries and responses we only need exclusive access in very
small chunks and the current implementation does it very well.

The performance bottleneck is purely server side. We will not have nor want
multiple connections to the same host.

The hand rolled proposal is built on top of independently useful building
blocks. The current connections are already done and work well. The
coordinator is going to be necessary for table affinity in any solution we
come across.

The ideal drop in solution would be something that:

1. keep track of established connections.
2. reconnect broken connections.

Number 2 is exactly what a supervisor is for. This is already an out of the
box solution. The only problem we need to solve is how a client find out
which connections currently are alive. I'm fairly certain which_children
will go a long way in solving that.

I'm going to take a very lightweight pass at this and see how difficult it
actually is.

On Thu, May 7, 2015, 8:26 AM Jason S. notifications@github.com wrote:

Their is an issue open for Ecto that solves this by "reserving" a worker
elixir-ecto/ecto#569 I'd be pretty strongly
against hand rolling because this kind of thing is full of pitfalls. We
could also ping @fishcakez and see if maybe his sbroker might be a stronger
use case here.

On Wed, May 6, 2015 at 11:15 PM, Peter Hamilton notifications@github.com
wrote:

Poolboy

Poolboy works by providing exclusive access to a given resource in a
"transaction".

Pros:

• Dynamic connection allocation is taken care of for you.

Cons:

• Connection affinity is required for cursors, which means a
  transaction may be required for the full length of the cursor. For long
  running change feeds, this will require one connection per changefeed.
• Pipelining is not utilized as each connection is checked out until a
  response is provided.

While my assumption was to use Poolboy, it seems that there are better
approaches available. There are too many tradeoffs to be made for not a
whole lot of benefit.
Hand rolled

An ideal connection pool will do the following:

1. Connect to N clients in a cluster.
2. Load balance queries between connections.
3. Route query to master replica for the target table.
4. If a connection dies, remove that connection from the rotation.
5. If a query fails because of a server connectivity issue, retry
   query on another host.

Initial design:

A supervisor tree will be over N connections and a coordinator.

Client processes send queries directly to the connection, just like we
currently do. In order to know which connection to contact, the client
can
request a connection from the coordinator. When the client is finished
with
the query it informs the coordinator (in order for the coordinator to
properly balance connections).

This would require zero changes to Exrethinkdb.Connection. The connection
would be dumb to pooling. We could provide a Exrethinkdb.ConnectionPool
module with run/next/close that wrap the logic for interacting with the
coordinator. The end API for the user would be the same. We will also
provide a use Exrethinkdb.ConnectionPool macro. Ideally you would
replace use
Exrethinkdb.Connection with use Exrethinkdb.ConnectionPool and your
application would still work perfectly.

Routing queries to proper replicas can be done fairly well in the
coordinator by looking at table("foo_table") |> config |> run. The
coordinator can do this on a timer (once every minute is probably
sufficient). It can then use this information to route the connection
properly. In the event of stale_reads, the coordinator can load balance
between non-master connections. We'll have to add table to the Query
struct, but that's pretty straightforward.

Exrethinkdb.ConnectionPool.run will report back to the coordinator the
success or failure of a query. The coordinator will also be monitoring
the
Connection processes. There will also be logic to retry on a different
server if a failure occurs (in the event of stale_reads. otherwise,
retrying won't be useful until a new master is selected).

I think this custom approach will be fairly simple. It will be designed
so
that failure leads to a worst case scenario of routing requests to the
less
optimal replica.

Reply to this email directly or view it on GitHub
#32.

—
Reply to this email directly or view it on GitHub
#32 (comment).

[jeregrine]

@hamiltop after talking with @ericmj and @fishcakez they've convinced me that maybe we don't need connection pooling and that I am probably wrong.

[hamiltop]

Table affinity is what's got me convinced I want to do some level of
connection management. I hacked together a lightweight naive pool and the
code is dead simple.

I skimmed the conversation in irc and I think your previous comment here
about this being a smarter client then something like postgres is accurate
and changes the discussion a bit.

I'll hopefully have my code up soon.

On Thu, May 7, 2015, 9:07 AM Jason S. notifications@github.com wrote:

@hamiltop https://github.com/hamiltop after talking with @ericmj
https://github.com/ericmj and @fishcakez https://github.com/fishcakez
they've convinced me that maybe we don't need connection pooling and that I
am probably wrong.

—
Reply to this email directly or view it on GitHub
#32 (comment).

[hamiltop]

https://github.com/hamiltop/exrethinkdb/blob/connection_pool/lib/exrethinkdb/connection/pool.ex

used like:

hosts = [[host: "localhost", port: 28015], [host: "localhost", port: 28015]]
{:ok, conn_pool} = Exrethinkdb.Connection.Pool.start_link([hosts: hosts])
conn = Exrethinkdb.Connection.Pool.get_connection(conn_pool)
Exrethinkdb.Query.table("people") |> Exrethinkdb.Connection.run conn
# alternatively, run from the conn_pool and it will pick a connection for you
Exrethinkdb.Query.table("people") |> Exrethinkdb.Connection.Pool.run conn_pool

It's naive, but accomplishes the basic goals. The optimizations would be in how get_connection works in order to provide table affinity.

As an already-implemented, cursors store the pid of their connection internally and will use that connection when next is called.

[Linicks]

@hamiltop, I really like the ideas that you've laid out for the hand rolled solution because it takes into account the unique characteristics of RethinkDB. Poolboy, and other technologies are going to make assumptions that are part of a different paradigm. I also like the fact that you were able to build a working version based on the code that's already been written. As @jeregrine mentioned, there are bound to be Gremlins lurking in the dark, but that seem to be the case for just about everything in this line of work. At this point, I can't think of a good reason not to run with your solution.

[hamiltop]

Some interesting results:

Three configurations:

1. Single local connection (single)
2. A Pool of 4 connections to localhost (pool)
3. A pool of 4 connections to localhost plus 1 connection to a remote host (cluster). The remote host has significant network latency (this is going from my laptop to my iMac over WiFi).

All of these were run with use_outdated: true. Without allowing outdated reads, the cluster mode has no real benefit if you aren't sharding your data. If you shard your data, the results are similar to use_outdated: true in that the clustering has benefit.

The first test was a non-indexed read in a table of 10k records (looking for one record), repeated 40 times across 40 Tasks for a total of 1600 queries and 1.6M reads (non-indexed reads count one read per record). This was very much CPU bound in rethinkdb.

Real results now
{:pool, {24016272, :ok}}
{:single, {23215845, :ok}}
{:cluster, {17805260, :ok}}
Again, reversed
{:cluster, {18998542, :ok}}
{:single, {23591027, :ok}}
{:pool, {23255329, :ok}}

The next test was for an indexed get in the same table, but repeated 4000 times across 40 Tasks for a total of 160000 queries and 160000 reads (one read per get). This was CPU bound on the elixir side.

Real results now
{:pool, {19722294, :ok}}
{:single, {20719733, :ok}}
{:cluster, {35041397, :ok}}
Again, reversed
{:cluster, {30579648, :ok}}
{:single, {20782334, :ok}}
{:pool, {20139856, :ok}}

Conclusions
For heavy non-indexed reads (or anything CPU bound in the DB) clustering is a benefit. For heavy indexed reads (or anything light on CPU in the DB) clustering slows things down (or at best doesn't really help much).

Additionally, fine tuning the pool is important. When I ran the cluster test with 1 local and 1 remote connection it was much slower than when I ran it with 4 local and 1 remote. This was in a very lopsided network, but I've encountered lopsided clusters aplenty in production environments.

I think the takeaway is that a developer will need to weight hosts in the pool according to their requirements. The configuration should be something like:

[
  [host: "db1", port: 28015, weight: 2],
  [host: "db2", port: 28015, weight: 3],
  [host: "db3", port: 28015, weight: 1]
]

Additionally, table affinity is definitely required. Hitting the connection that isn't the master for the table was costly.

[Linicks]

The results are exactly what I would expect them to be given the parameters you described. In my mind this seems to indicate that initial design is working properly. This will add the flexibility needed to get the most out of RethinkDB and the application. A couple of things that I was thinking about were pools that had equally weighted hosts. For example, if you have a RethinkDB cluster in data-center A, and another RethinkDB Cluster in data-center B, it seems logical to have the following configuration:

[
[host: "dca_db1", port: 28015, weight: 0],
[host: "dca_db2", port: 28015, weight: 0],
[host: "dca_db3", port: 28015, weight: 0]
[host: "dcb_db1", port: 28015, weight: 1],
[host: "dcb_db2", port: 28015, weight: 1],
[host: "dcb_db3", port: 28015, weight: 1]
]

This would allow the application servers in data-center A to utilize the local network for best performance, but fall back to the cluster in data-center B. Given that they are equally weighted in each group, I would expect that the load would be balanced among them.

I have been kicking around the idea of adding tags that could be used to specify things like racks, data-centers, etc, but the weight parameter may cover most cases. At this point I'm not sure if they would help or hurt, but I thought the idea was at least worth mentioning.

--Nick

[hamiltop]

weight: 0 would need to be a special case, as otherwise weights are
traditionally applied as P(use this host) = (weight of host)/sum(weights)
which in the case of weight 0 would be 0% probability. We could probably
special case it so that is sum(weights) is 0 the we just round robin.

I did find a bug in some code not pushed yet. :random.uniform has a state
local to each process. The current code (my new code has not been pushed
yet) resulted in all processes using the same round robin ordering of
connections. The tests were very useful in exposing that.

On Fri, May 8, 2015 at 1:29 PM Nick notifications@github.com wrote:

The results are exactly what I would expect them to be given the
parameters you described. In my mind this seems to indicate that initial
design is working properly. This will add the flexibility needed to get the
most out of RethinkDB and the application. A couple of things that I was
thinking about were pools that had equally weighted hosts. For example, if
you have a RethinkDB cluster in data-center A, and another RethinkDB
Cluster in data-center B, it seems logical to have the following
configuration:

[
[host: "dca_db1", port: 28015, weight: 0],
[host: "dca_db2", port: 28015, weight: 0],
[host: "dca_db3", port: 28015, weight: 0]
[host: "dcb_db1", port: 28015, weight: 1],
[host: "dcb_db2", port: 28015, weight: 1],
[host: "dcb_db3", port: 28015, weight: 1]
]

This would allow the application servers in data-center A to utilize the
local network for best performance, but fall back to the cluster in
data-center B. Given that they are equally weighted in each group, I would
expect that the load would be balanced among them.

I have been kicking around the idea of adding tags that could be used to
specify things like racks, data-centers, etc, but the weight parameter may
cover most cases. At this point I'm not sure if they would help or hurt,
but I thought the idea was at least worth mentioning.

--Nick

—
Reply to this email directly or view it on GitHub
#32 (comment)
.

[fishcakez]

You have a couple of issues regarding failure handling with the proposed pooling method:

The case where all connections are unavailable isn't handled. This could happen if all connections are :restarting, in which case a client will crash when generating the random integer. Perhaps that is desired.

A supervisor blocks while starting a child. In the current implementation connect and handshake occur in the init/1 callback. This means that the supervisor/the pool is blocked and all client processes requesting a worker are blocked too while a single connection process starts.

If one of the rethinkdb servers is down or there is a issue with connecting/sending data, a connection process will fail to connect. This will either result in the supervisor blocking for a significant time while a connection process tries to start (as above) or more likely the supervisor will shutdown because the max_restarts limit was reached because a connection process keeps failing to start in rapid succession.

Finally the poolling strategy does not provide any backpressure or load shedding to client processes beyond the connection processes' message queues. Designing around a minimal number of connection processes that pipeline requests makes this is a potential bottleneck.

[hamiltop]

Thanks for chiming in @fishcakez

The case for "unable to acquire a connection" has not been addressed at any level yet, but either a "let it crash" philosophy or responding with {:error, msg} should suffice for both the individual connection case and the pool case.

To check my understanding, the init function of a GenServer is run as part of GenServer.start_link and therefore it causes the supervisor to block. Is it a common pattern to send oneself a :connect message to be processed immediately after init? How is this otherwise handled?

How is the max_restarts issue (where one buggy child kills all the other healthy children as it brings down the supervisor) solved elsewhere? By not using supervision?

Backpressure is applied to individual clients through GenServer.call. The client process will block until it receives a response. This does nothing to address an unbounded number of client processes. It also does not deal with load shedding, as stated. Would a simple solution be for an individual connection to keep track of active clients and respond :too_busy? The client should then try another connection or in the case of not using a connection pool just fail. Change feeds would need to be accounted for differently because they don't have the same resource footprint as active connections.

While I'm exploring the idea of a custom pool with table affinity, I'm committed to making Connection compatible with existing connection pooling solutions like poolboy. Not using pipelining isn't the end of the world, and this is how it is done in other languages. This custom pooling is an optimization. Perhaps there's a hybrid solution that can be found.