<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def ready(self, cluster):
 """ Get a list of nodes whose partitions are ready to be sent, and the earliest time at which any non-sendable partition will be ready; Also return the flag for whether there are any unknown leaders for the accumulated partition batches. A destination node is ready to send if: * There is at least one partition that is not backing off its send * and those partitions are not muted (to prevent reordering if max_in_flight_requests_per_connection is set to 1) * and any of the following are true: * The record set is full * The record set has sat in the accumulator for at least linger_ms milliseconds * The accumulator is out of memory and threads are blocking waiting for data (in this case all partitions are immediately considered ready). * The accumulator has been closed Arguments: cluster (ClusterMetadata):
 Returns: tuple: ready_nodes (set):
 node_ids that have ready batches next_ready_check (float):
 secs until next ready after backoff unknown_leaders_exist (bool):
 True if metadata refresh needed """

        ready_nodes = set()
        next_ready_check = 9999999.99
        unknown_leaders_exist = False
        now = time.time()

        exhausted = bool(self._free.queued() > 0)
        # several threads are accessing self._batches -- to simplify
        # concurrent access, we iterate over a snapshot of partitions
        # and lock each partition separately as needed
        partitions = list(self._batches.keys())
        for tp in partitions:
            leader = cluster.leader_for_partition(tp)
            if leader is None or leader == -1:
                unknown_leaders_exist = True
                continue
            elif leader in ready_nodes:
                continue
            elif tp in self.muted:
                continue

            with self._tp_locks[tp]:
                dq = self._batches[tp]
                if not dq:
                    continue
                batch = dq[0]
                retry_backoff = self.config['retry_backoff_ms'] / 1000.0
                linger = self.config['linger_ms'] / 1000.0
                backing_off = bool(batch.attempts > 0 and
                                   batch.last_attempt + retry_backoff > now)
                waited_time = now - batch.last_attempt
                time_to_wait = retry_backoff if backing_off else linger
                time_left = max(time_to_wait - waited_time, 0)
                full = bool(len(dq) > 1 or batch.records.is_full())
                expired = bool(waited_time >= time_to_wait)

                sendable = (full or expired or exhausted or self._closed or
                            self._flush_in_progress())

                if sendable and not backing_off:
                    ready_nodes.add(leader)
                else:
                    # Note that this results in a conservative estimate since
                    # an un-sendable partition may have a leader that will
                    # later be found to have sendable data. However, this is
                    # good enough since we'll just wake up and then sleep again
                    # for the remaining time.
                    next_ready_check = min(time_left, next_ready_check)

        return ready_nodes, next_ready_check, unknown_leaders_exist