1212
1313 reusability
1414"""
15+ import socket
1516import logging
1617import selectors
1718
@@ -45,11 +46,19 @@ class UpstreamConnectionPool(Work[TcpServerConnection]):
4546 A separate pool is maintained for each upstream server.
4647 So internally, it's a pool of pools.
4748
48- TODO: Listen for read events from the connections
49- to remove them from the pool when peer closes the
50- connection. This can also be achieved lazily by
51- the pool users. Example, if acquired connection
52- is stale, reacquire.
49+ Internal data structure maintains references to connection objects
50+ that pool owns or has borrowed. Borrowed connections are marked as
51+ NOT reusable.
52+
53+ For reusable connections only, pool listens for read events
54+ to detect broken connections. This can happen if pool has opened
55+ a connection, which was never used and eventually reaches
56+ upstream server timeout limit.
57+
58+ When a borrowed connection is returned back to the pool,
59+ the connection is marked as reusable again. However, if
60+ returned connection has already been closed, it is removed
61+ from the internal data structure.
5362
5463 TODO: Ideally, `UpstreamConnectionPool` must be shared across
5564 all cores to make SSL session cache to also work
@@ -60,29 +69,25 @@ class UpstreamConnectionPool(Work[TcpServerConnection]):
6069 session cache, session ticket, abbr TLS handshake
6170 and other necessary features to make it work.
6271
63- NOTE: However, for all HTTP only connections, `UpstreamConnectionPool`
64- can be used to save upon connection setup time and
65- speed-up performance of requests.
72+ NOTE: However, currently for all HTTP only upstream connections,
73+ `UpstreamConnectionPool` can be used to remove slow starts.
6674 """
6775
6876 def __init__ (self ) -> None :
69- # Pools of connection per upstream server
7077 self .connections : Dict [int , TcpServerConnection ] = {}
7178 self .pools : Dict [Tuple [str , int ], Set [TcpServerConnection ]] = {}
7279
7380 def add (self , addr : Tuple [str , int ]) -> TcpServerConnection :
74- # Create new connection
81+ """Creates and add a new connection to the pool."""
7582 new_conn = TcpServerConnection (addr [0 ], addr [1 ])
7683 new_conn .connect ()
77- if addr not in self .pools :
78- self .pools [addr ] = set ()
79- self .pools [addr ].add (new_conn )
80- self .connections [new_conn .connection .fileno ()] = new_conn
84+ self ._add (new_conn )
8185 return new_conn
8286
8387 def acquire (self , addr : Tuple [str , int ]) -> Tuple [bool , TcpServerConnection ]:
84- """Returns a connection for use with the server."""
85- # Return a reusable connection if available
88+ """Returns a reusable connection from the pool.
89+
90+ If none exists, will create and return a new connection."""
8691 if addr in self .pools :
8792 for old_conn in self .pools [addr ]:
8893 if old_conn .is_reusable ():
@@ -102,40 +107,63 @@ def acquire(self, addr: Tuple[str, int]) -> Tuple[bool, TcpServerConnection]:
102107 return True , new_conn
103108
104109 def release (self , conn : TcpServerConnection ) -> None :
105- """Release the connection.
110+ """Release a previously acquired connection.
106111
107112 If the connection has not been closed,
108113 then it will be retained in the pool for reusability.
109114 """
115+ assert not conn .is_reusable ()
110116 if conn .closed :
111117 logger .debug (
112118 'Removing connection#{2} from pool from upstream {0}:{1}' .format (
113119 conn .addr [0 ], conn .addr [1 ], id (conn ),
114120 ),
115121 )
116- self .pools [ conn .addr ]. remove ( conn )
122+ self ._remove ( conn .connection . fileno () )
117123 else :
118124 logger .debug (
119125 'Retaining connection#{2} to upstream {0}:{1}' .format (
120126 conn .addr [0 ], conn .addr [1 ], id (conn ),
121127 ),
122128 )
123- assert not conn .is_reusable ()
124129 # Reset for reusability
125130 conn .reset ()
126131
127132 async def get_events (self ) -> Dict [int , int ]:
133+ """Returns read event flag for all reusable connections in the pool."""
128134 events = {}
129135 for connections in self .pools .values ():
130136 for conn in connections :
131- events [conn .connection .fileno ()] = selectors .EVENT_READ
137+ if conn .is_reusable ():
138+ events [conn .connection .fileno ()] = selectors .EVENT_READ
132139 return events
133140
134141 async def handle_events (self , readables : Readables , _writables : Writables ) -> bool :
135- for r in readables :
142+ """Removes reusable connection from the pool.
143+
144+ When pool is the owner of connection, we don't expect a read event from upstream
145+ server. A read event means either upstream closed the connection or connection
146+ has somehow reached an illegal state e.g. upstream sending data for previous
147+ connection acquisition lifecycle."""
148+ for fileno in readables :
136149 if TYPE_CHECKING :
137- assert isinstance (r , int )
138- conn = self .connections [r ]
139- self .pools [conn .addr ].remove (conn )
140- del self .connections [r ]
150+ assert isinstance (fileno , int )
151+ logger .debug ('Upstream fd#{0} is read ready' .format (fileno ))
152+ self ._remove (fileno )
141153 return False
154+
155+ def _add (self , conn : TcpServerConnection ) -> None :
156+ """Adds a new connection to internal data structure."""
157+ if conn .addr not in self .pools :
158+ self .pools [conn .addr ] = set ()
159+ self .pools [conn .addr ].add (conn )
160+ self .connections [conn .connection .fileno ()] = conn
161+
162+ def _remove (self , fileno : int ) -> None :
163+ """Remove a connection by descriptor from the internal data structure."""
164+ conn = self .connections [fileno ]
165+ logger .debug ('Removing conn#{0} from pool' .format (id (conn )))
166+ conn .connection .shutdown (socket .SHUT_WR )
167+ conn .close ()
168+ self .pools [conn .addr ].remove (conn )
169+ del self .connections [fileno ]
![pre-commit-ci[bot]](https://avatars.githubusercontent.com/in/68672?v=4&size=40){kind=avatar}
0 commit comments