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

orinium_browser/platform/network/
mod.rs
