median

    def find_median(self):
        """
        TASK 3: Find the median using a clean 3-phase distributed quickselect.
        This version is corrected with a simplified and robust aggregation logic.
        """
        # === PHASE 1: SETUP ===
        local_size = len(self.data)
        if self.worker_id != 0:
            self.send(0, {'type': 'size', 'value': local_size})
        yield

        if self.worker_id == 0:
            total_size = local_size
            for msg in self.recv():
                total_size += msg['value']
            if total_size == 0:
                for i in range(self.num_workers):
                    self.send(i, {'type': 'result', 'value': None})
            else:
                target_rank = (total_size - 1) // 2
                for i in range(self.num_workers):
                    self.send(i, {'type': 'setup', 'target_rank': target_rank})
        yield

        setup_msg = self.recv()[0]
        if setup_msg.get('type') == 'result':
            return setup_msg['value'] if self.worker_id == 0 else None

        current_target = setup_msg['target_rank']
        current_data = list(self.data)
        pivot = None

        # === PHASE 2: QUICKSELECT LOOP ===
        while True:
            # --- Round 1: Propose, Select, and Broadcast Pivot ---
            if current_data:
                self.send(0, {'type': 'pivot_candidate', 'value': random.choice(current_data)})
            yield

            if self.worker_id == 0:
                candidates = [msg['value'] for msg in self.recv()]
                if not candidates:
                    raise RuntimeError("Median not found, but no data remains.")

                pivot = random.choice(candidates)
                for i in range(self.num_workers):
                    self.send(i, {'type': 'pivot', 'value': pivot})
            yield

            # All workers receive the pivot.
            if self.worker_id != 0:
                messages = self.recv()
                if messages:
                    pivot = messages[0]['value']
            else:
                self.recv() # Worker 0 just clears the message it sent to itself.

            # --- Round 2: Partition and Count ---
            less_than_count = sum(1 for x in current_data if x < pivot)
            equal_to_count = sum(1 for x in current_data if x == pivot)
            
            # THE FIX: Only workers 1-N send counts. Worker 0 will use its local values directly.
            if self.worker_id != 0:
                self.send(0, {'type': 'counts', 'lt': less_than_count, 'eq': equal_to_count})
            yield

            # --- Round 3: Decide and Broadcast Decision ---
            if self.worker_id == 0:
                # Start totals with worker 0's own, direct counts.
                total_lt = less_than_count
                total_eq = equal_to_count

                # Add counts received from all other workers.
                for msg in self.recv():
                    total_lt += msg['lt']
                    total_eq += msg['eq']

                if current_target < total_lt:
                    decision = 'keep_less'
                    new_target = current_target
                elif current_target < total_lt + total_eq:
                    decision = 'found'
                    new_target = None
                else:
                    decision = 'keep_greater'
                    new_target = current_target - (total_lt + total_eq)

                broadcast_msg = {
                    'decision': decision,
                    'new_target': new_target,
                    'result': pivot if decision == 'found' else None
                }
                for i in range(self.num_workers):
                    self.send(i, broadcast_msg)
            yield

            # ALL workers get the canonical decision message from the queue.
            decision_msg = self.recv()[0]

            decision = decision_msg['decision']
            if decision == 'found':
                return decision_msg['result'] if self.worker_id == 0 else None

            current_target = decision_msg['new_target']
            if decision == 'keep_less':
                current_data = [x for x in current_data if x < pivot]
            elif decision == 'keep_greater':
                current_data = [x for x in current_data if x > pivot]