Method run_worker

src/storage/src/server.rs:123–204

(
        &self,
        timely_worker: &mut TimelyWorker,
        client_rx: mpsc::UnboundedReceiver<(
            Uuid,
            mpsc::UnboundedReceiver<StorageCommand>,
            mpsc::Unbound

    const NAME: &str = "storage";

    fn run_worker(
        &self,
        timely_worker: &mut TimelyWorker,
        client_rx: mpsc::UnboundedReceiver<(
            Uuid,
            mpsc::UnboundedReceiver<StorageCommand>,
            mpsc::UnboundedSender<StorageResponse>,
        )>,
    ) {
        // Register a timely logger that forwards events to the compute logging dataflow.
        // Assign by local worker index so storage worker x matches compute worker x.
        let local_index = timely_worker.index() % self.workers_per_process;
        let writer = self.timely_log_writers.lock().unwrap()[local_index].take();
        if let Some(writer) = writer {
            use timely::dataflow::operators::capture::{Event, EventPusher};
            use timely::logging::TimelyEventBuilder;

            // We use an approach similar to compute's logging: wrap the writer in
            // a BatchLogger that translates Logger callbacks into Event pushes,
            // then register the Logger with timely's log_register.
            let interval_ms = 1000u128; // 1 second batching interval
            let mut time_ms = mz_repr::Timestamp::from(0u64);
            let mut event_pusher = writer;
            let now = std::time::Instant::now();
            let start_offset = std::time::SystemTime::now()
                .duration_since(std::time::SystemTime::UNIX_EPOCH)
                .expect("Failed to get duration since Unix epoch");

            let logger = timely::logging_core::Logger::<TimelyEventBuilder>::new(
                now,
                start_offset,
                move |time: &std::time::Duration,
                      data: &mut Option<Vec<(std::time::Duration, TimelyEvent)>>| {
                    if let Some(mut data) = data.take() {
                        // Filter park/unpark events and remap IDs before handing events
                        // off to compute. Compute's park tracking assumes a single
                        // timely runtime; mixing in storage's park events would break
                        // it. Remapping ensures storage operator/channel IDs don't
                        // collide with compute's.
                        data.retain_mut(|(_, event)| {
                            if matches!(event, TimelyEvent::Park(_)) {
                                return false;
                            }
                            remap_timely_event_ids(event);
                            true
                        });
                        event_pusher.push(Event::Messages(time_ms, data));
                    } else {
                        // Advance progress.
                        let new_time_ms: u64 = (((time.as_millis() / interval_ms) + 1)
                            * interval_ms)
                            .try_into()
                            .expect("must fit");
                        let new_time_ms = mz_repr::Timestamp::from(new_time_ms);
                        if time_ms < new_time_ms {
                            event_pusher
                                .push(Event::Progress(vec![(new_time_ms, 1), (time_ms, -1)]));
                            time_ms = new_time_ms;