orinium_browser/platform/network/
mod.rs

1//! Network processing module, providing HTTP communication, cache, and cookie management.
2
3pub mod cache;
4pub mod config;
5pub mod cookie_store;
6mod core;
7pub mod error;
8pub mod sender_pool;
9
10pub use cache::Cache;
11pub use config::NetworkConfig;
12pub use cookie_store::CookieStore;
13pub use core::{Response, StatusCode};
14pub use error::NetworkError;
15pub use hyper::http::Request;
16use ipc_channel::IpcError;
17pub use sender_pool::HostKey;
18pub use sender_pool::{HttpSender, SenderPool};
19
20use serde::{Deserialize, Serialize};
21
22use core::{AsyncNetworkCore, SharedNetState};
23
24use crate::ParentChannels;
25use crate::engine::background_worker::BackgroundWorker;
26use ipc_channel::ipc::{IpcOneShotServer, IpcReceiver, IpcSender};
27use std::cell::Cell;
28use std::sync::Arc;
29use std::sync::atomic::{AtomicUsize, Ordering};
30use std::{env, io, process};
31
32#[derive(Deserialize, Serialize)]
33pub enum NetworkCommand {
34    Fetch {
35        request: NetworkRequest,
36        msg_id: usize,
37    },
38    SetConfig(NetworkConfig),
39    ClearCache,
40}
41
42/// Serializable HTTP request passed to the network process.
43#[derive(Clone, Debug, Deserialize, Serialize)]
44pub struct NetworkRequest {
45    pub url: String,
46    pub method: String,
47    pub headers: Vec<(String, String)>,
48    pub body: Vec<u8>,
49}
50
51impl NetworkRequest {
52    pub fn get(url: impl Into<String>) -> Self {
53        Self {
54            url: url.into(),
55            method: "GET".to_string(),
56            headers: Vec::new(),
57            body: Vec::new(),
58        }
59    }
60}
61
62#[derive(Deserialize, Serialize)]
63pub struct NetworkMessage {
64    pub msg_id: usize,
65    pub response: Result<Response, NetworkError>,
66}
67
68pub struct NetworkCore {
69    cmd_tx: IpcSender<NetworkCommand>,
70    msg_rx: IpcReceiver<NetworkMessage>, // UI スレッド用
71}
72
73impl Default for NetworkCore {
74    fn default() -> Self {
75        Self::new().unwrap()
76    }
77}
78
79impl NetworkCore {
80    pub fn new() -> Result<Self, io::Error> {
81        let (server, server_name) =
82            IpcOneShotServer::<ParentChannels<NetworkCommand, NetworkMessage>>::new()?;
83
84        process::Command::new(env::current_exe()?)
85            .arg("--child")
86            .arg(&server_name)
87            .arg("--type=network")
88            .spawn()?;
89
90        let (_, channels) = server.accept().unwrap();
91
92        Ok(Self {
93            cmd_tx: channels.cmd_tx,
94            msg_rx: channels.msg_rx,
95        })
96    }
97
98    pub fn set_network_config(&self, cfg: NetworkConfig) {
99        let _ = self.cmd_tx.send(NetworkCommand::SetConfig(cfg));
100    }
101
102    /// Clears all cached responses in the network process.
103    pub fn clear_cache(&self) {
104        let _ = self.cmd_tx.send(NetworkCommand::ClearCache);
105    }
106
107    /// 非同期送信のみ。結果は try_receive で取得
108    pub fn fetch_async(&self, url: String, msg_id: usize) {
109        self.fetch_request_async(NetworkRequest::get(url), msg_id);
110    }
111
112    pub fn fetch_request_async(&self, request: NetworkRequest, msg_id: usize) {
113        let _ = self.cmd_tx.send(NetworkCommand::Fetch { request, msg_id });
114    }
115
116    /// UIスレッドから呼ぶ: 完了しているメッセージを取り込む
117    pub fn try_receive(&self) -> Vec<NetworkMessage> {
118        let mut msgs = Vec::new();
119        while let Ok(msg) = self.msg_rx.try_recv() {
120            log::info!(
121                target: "network",
122                "return message for msg_id={}",
123                msg.msg_id
124            );
125            msgs.push(msg);
126        }
127        msgs
128    }
129
130    pub fn fetch_blocking(&self, url: &str) -> Result<Response, NetworkError> {
131        self.fetch_async(url.to_string(), 0);
132        loop {
133            if let Some(v) = self.try_receive().into_iter().next() {
134                return v.response;
135            }
136            std::thread::yield_now();
137        }
138    }
139}
140
141/// Fetch pool size: network work is IO-bound and benefits from some
142/// over-subscription, but each worker holds its own tokio runtime, so the
143/// count is capped well above the CPU count without growing unbounded.
144fn network_worker_count() -> usize {
145    std::thread::available_parallelism()
146        .map(|parallelism| parallelism.get())
147        .unwrap_or(2)
148        .clamp(1, 6)
149}
150
151static NEXT_FETCH_WORKER_ID: AtomicUsize = AtomicUsize::new(0);
152
153thread_local! {
154    static FETCH_WORKER_ID: Cell<Option<usize>> = const { Cell::new(None) };
155}
156
157/// `[N]` prefix identifying the fetch pool worker emitting a log line.
158///
159/// Empty on threads outside the pool (the IPC loop), so those lines stay
160/// unprefixed.
161fn fetch_worker_tag() -> String {
162    FETCH_WORKER_ID.with(|slot| match slot.get() {
163        Some(id) => format!("[{id}] "),
164        None => String::new(),
165    })
166}
167
168/// ネットワークプロセスエントリ
169///
170/// Fetches run on a [`BackgroundWorker`] pool so one slow request no longer
171/// blocks every other tab or subresource. Config and cache commands stay on
172/// this thread: they mutate the state shared by all workers.
173pub fn network_main(rx: IpcReceiver<NetworkCommand>, tx: IpcSender<NetworkMessage>) -> ! {
174    let shared = Arc::new(SharedNetState::new());
175
176    let worker = BackgroundWorker::new_with_init(
177        network_worker_count(),
178        {
179            let shared = Arc::clone(&shared);
180            move || {
181                let id = NEXT_FETCH_WORKER_ID.fetch_add(1, Ordering::SeqCst);
182                FETCH_WORKER_ID.with(|slot| slot.set(Some(id)));
183                log::info!(target: "network", "[{id}] fetch worker started");
184                AsyncNetworkCore::new(Arc::clone(&shared))
185            }
186        },
187        move |core, (request, msg_id): (NetworkRequest, usize)| {
188            let started = std::time::Instant::now();
189            let response = core.fetch_request_blocking(&request);
190            match &response {
191                Ok(res) => log::info!(
192                    target: "network",
193                    "{}fetch completed: msg_id={} url={} status={} body={}B took={:?}",
194                    fetch_worker_tag(),
195                    msg_id,
196                    res.url,
197                    res.status.as_u16(),
198                    res.body.len(),
199                    started.elapsed(),
200                ),
201                Err(err) => log::warn!(
202                    target: "network",
203                    "{}fetch failed: msg_id={} error={} took={:?}",
204                    fetch_worker_tag(),
205                    msg_id,
206                    err,
207                    started.elapsed(),
208                ),
209            }
210            let _ = tx.send(NetworkMessage { msg_id, response });
211        },
212    );
213
214    while let Ok(cmd) = rx.recv() {
215        match cmd {
216            NetworkCommand::SetConfig(cfg) => shared.set_network_config(cfg),
217            NetworkCommand::ClearCache => {
218                shared.clear_cache();
219                log::info!(target: "network", "cache cleared");
220            }
221            NetworkCommand::Fetch { request, msg_id } => {
222                log::info!(
223                    target: "network",
224                    "fetch dispatched: msg_id={} url={} method={}",
225                    msg_id,
226                    request.url,
227                    request.method
228                );
229                worker.send((request, msg_id));
230            }
231        }
232    }
233
234    drop(worker);
235
236    let err = rx.recv().err().unwrap();
237
238    if matches!(err, IpcError::Disconnected) {
239        log::info!(target: "network", "IPC channel closed, exiting normally.");
240        std::process::exit(0)
241    } else {
242        panic!("IPC channel unexpectedly closed: {err}")
243    }
244}
orinium_browser/platform/network/mod.rs

orinium_browser/platform/network/
mod.rs
