Clean closed connections.

        Args:
            force: Clean connected and idle connections too.

        Coroutine.
328295
def clean(self, force: bool=False):
        '''
        '''
        with (yield from self._lock):
            for connection in tuple(self.ready):
                if force or connection.closed():
                    connection.close()
                    self.ready.remove(connection)
640155
def _reset(self, force=False):
        
        if not self._closed and (force or self._transaction):
            try:
                self.rollback()
            except Exception:
                pass
1084812
def _clear(reason, idle_pool, using_pool, channel_pool, conn_id):
        
        with _lock():
            try:
                idle_pool.pop(conn_id)
                logger.info('a connection lost when not using')
            except KeyError:
                if using_pool.pop(conn_id, None):
                    logger.warn('connection lost when using, should be handled later')
                    return reason
            finally:
                channel_pool.pop(conn_id, None)
1361050
def sync_close_connection(self):
        """"""
        loop = asyncio.get_event_loop()
        task = loop.create_task(self.close_connection())
        loop.run_until_complete(task)
863694
async def reset(self, force = True, connmark = None):
        '''
        
        '''
        if connmark is None:
            connmark = self.connmark
        self.scheduler.emergesend(ConnectionControlEvent(self, ConnectionControlEvent.RESET, force, connmark))
738933
def flush(self):
        # type: () -> None
        
        if self._closed:
            raise IOError('tried to flush closed connection')
        assert self._parentout is not None
        self._parentout.flush()
1030220
def Cleanup(self):
		
		self.lock.acquire()
		try:
			for bucket in self.connections.values():
				try:
					for conn in bucket:
						conn.lock()
						try:
							open = conn.TestConnection(forceCheck=True)
							if open is True:
								conn.commit()
							else:
								#Remove the connection from the pool. Its dead better of recreating it.
								index = bucket.index(conn)
								del bucket[index]
							conn.release()
						except Exception:
							conn.release()
				except Exception:
					pass
		finally:
			self.lock.release()
1146792
def _cleanup_fd(self, fd, close=False):
        
        self._ioloop.remove_handler(fd)
        if fd in self._connections:
            try:
                self._pool_manager.free(self.pid, self._connections[fd])
            except pool.ConnectionNotFoundError:
                pass
            if close:
                self._connections[fd].close()
            del self._connections[fd]
        if fd in self._futures:
            del self._futures[fd]
972517
def maintainConnections(self, force=False):
        
        now = time.perf_counter()
        if now < self.nextCheck and not force:
            return False
        self.nextCheck = now + (self.config.RETRY_TIMEOUT_NOT_RESTRICTED
                                if self.isKeySharing
                                else self.config.RETRY_TIMEOUT_RESTRICTED)
        missing = self.connectToMissing()
        self.retryDisconnected(exclude=missing)
        logger.trace("{} next check for retries in {:.2f} seconds"
                     .format(self, self.nextCheck - now))
        return True
1301550
@VisibleForTesting
    @Synchronized
    public void pruneUnusedConnections() {
        for (List<Connection> connections : connectionMap.values()) {
            for (Iterator<Connection> iterator = connections.iterator(); iterator.hasNext();) {
                Connection connection = iterator.next();
                if (isUnused(connection)) {
                    connection.getFlowHandler().close();
                    iterator.remove();
                }
            }
        }
    }
398040
def close(self):
        
        if self._closed:
            return
        try:
            with self.lock:
                if not self._closed:
                    self._closed = True
                    for address in list(self.connections):
                        self.remove(address)
        except TypeError as e:
            pass
28797
async def clean(self):
        
        for x in self._running_asyncio_tasks:
            x.cancel()
        self._container_had_error = set()
        self._watching = set()
        self._running_asyncio_tasks = set()
631697
def shutdown(force=false)
      status(:closed)
      args = []
      args.push(:force) if force
      @logger.info("Pool is now shutting down #{force ? 'using force' : ''}")
      @queue.clear if force
      @queue.wait_empty
      while(t = @threads.pop) do
        t.stop(*args)
      end
      unless(force)
        flush
        @threads.each{|x|x.join}
      end
      nil
    end
1364410
def _force_disconnect_action(self, action):
        

        conn_key = action.data['id']
        if self._get_connection_state(conn_key) == self.Disconnected:
            return

        data = self._get_connection(conn_key)

        # If there are any operations in progress, cancel them cleanly
        if data['state'] == self.Connecting:
            callback = data['callback']
            callback(False, 'Unexpected disconnection')
        elif data['state'] == self.Disconnecting:
            callback = data['callback']
            callback(True, None)
        elif data['state'] == self.InProgress:
            callback = data['callback']
            if data['microstate'] == 'rpc':
                callback(False, 'Unexpected disconnection', None, None)
            elif data['microstate'] == 'open_interface':
                callback(False, 'Unexpected disconnection')
            elif data['microstate'] == 'close_interface':
                callback(False, 'Unexpected disconnection')

        int_id = data['int_id']
        conn_id = data['conn_id']
        del self._connections[conn_id]
        del self._int_connections[int_id]
717315
def _close_connections(self, connection=None, timeout=5):
        
        all = []
        if connection:
            waiter = connection.event('connection_lost').waiter()
            if waiter:
                all.append(waiter)
                connection.close()
        else:
            connections = list(self._concurrent_connections)
            self._concurrent_connections = set()
            for connection in connections:
                waiter = connection.event('connection_lost').waiter()
                if waiter:
                    all.append(waiter)
                    connection.close()
        if all:
            self.logger.info('%s closing %d connections', self, len(all))
            return asyncio.wait(all, timeout=timeout, loop=self._loop)
846175
def _on_closed(self):
        """"""
        LOGGER.error('Redis connection closed')
        self.connected = False
        self._on_close()
        self._stream = None
712014
def clean(self):
        """"""
        connection = self._backend._get_connection()
        self._backend.clean(connection)