opendal_layer_hotpath/
lib.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18#![doc = include_str!("../README.md")]
19#![cfg_attr(docsrs, feature(doc_cfg))]
20#![cfg_attr(docsrs, doc(auto_cfg))]
21#![deny(missing_docs)]
22use std::pin::Pin;
23use std::sync::Arc;
24use std::task::Context;
25use std::task::Poll;
26
27use futures::Stream;
28use futures::StreamExt;
29use opendal_core::raw::*;
30use opendal_core::*;
31
32const LABEL_CREATE_DIR: &str = "opendal.create_dir";
33const LABEL_READ: &str = "opendal.read";
34const LABEL_RENAME: &str = "opendal.rename";
35const LABEL_RESTORE: &str = "opendal.restore";
36const LABEL_STAT: &str = "opendal.stat";
37const LABEL_PRESIGN: &str = "opendal.presign";
38
39const LABEL_READER_READ: &str = "opendal.reader.read";
40const LABEL_WRITER_WRITE: &str = "opendal.writer.write";
41const LABEL_WRITER_CLOSE: &str = "opendal.writer.close";
42const LABEL_WRITER_ABORT: &str = "opendal.writer.abort";
43const LABEL_LISTER_NEXT: &str = "opendal.lister.next";
44const LABEL_DELETER_DELETE: &str = "opendal.deleter.delete";
45const LABEL_DELETER_CLOSE: &str = "opendal.deleter.close";
46const LABEL_COPIER_NEXT: &str = "opendal.copier.next";
47const LABEL_COPIER_CLOSE: &str = "opendal.copier.close";
48const LABEL_COPIER_ABORT: &str = "opendal.copier.abort";
49const LABEL_HTTP_FETCH: &str = "opendal.http.fetch";
50const LABEL_HTTP_BODY_POLL: &str = "opendal.http.body.poll";
51
52/// `HotpathLayer` profiles every operation with
53/// [hotpath](https://docs.rs/hotpath/).
54///
55/// # Notes
56///
57/// When `hotpath` profiling is enabled, initialize a guard via
58/// [`hotpath::HotpathGuardBuilder`] or `#[hotpath::main]` before running
59/// operations. Otherwise, hotpath will panic on the first measurement.
60///
61/// # Examples
62///
63/// ```no_run
64/// # use opendal_core::services;
65/// # use opendal_core::Operator;
66/// # use opendal_core::Result;
67/// # use opendal_layer_hotpath::HotpathLayer;
68/// #
69/// # #[tokio::main]
70/// # async fn main() -> Result<()> {
71/// let _guard = hotpath::HotpathGuardBuilder::new("opendal").build();
72/// let op = Operator::new(services::Memory::default())?
73///     .layer(HotpathLayer::new());
74/// op.write("test", "hello").await?;
75/// # Ok(())
76/// # }
77/// ```
78#[derive(Clone, Debug, Default)]
79#[non_exhaustive]
80pub struct HotpathLayer {}
81
82impl HotpathLayer {
83    /// Create a new [`HotpathLayer`].
84    pub fn new() -> Self {
85        Self::default()
86    }
87}
88
89impl Layer for HotpathLayer {
90    fn apply_service(&self, inner: Servicer) -> Servicer {
91        Arc::new(self.layer(inner))
92    }
93
94    fn apply_context(&self, _srv: Servicer, inner: OperationContext) -> OperationContext {
95        let transport = HttpTransporter::new(HotpathHttpTransport {
96            inner: inner.http_transport().clone(),
97        });
98        inner.with_http_transport(transport)
99    }
100}
101
102impl HotpathLayer {
103    fn layer(&self, inner: Servicer) -> HotpathAccessor {
104        HotpathAccessor { inner }
105    }
106}
107
108#[doc(hidden)]
109#[derive(Debug)]
110pub struct HotpathAccessor {
111    inner: Servicer,
112}
113
114impl Service for HotpathAccessor {
115    type Reader = HotpathWrapper<oio::Reader>;
116    type Writer = HotpathWrapper<oio::Writer>;
117    type Lister = HotpathWrapper<oio::Lister>;
118    type Deleter = HotpathWrapper<oio::Deleter>;
119    type Copier = HotpathWrapper<oio::Copier>;
120    type Composer = oio::Composer;
121
122    fn info(&self) -> ServiceInfo {
123        self.inner.info()
124    }
125
126    fn capability(&self) -> Capability {
127        self.inner.capability()
128    }
129
130    fn compose(&self, ctx: &OperationContext, to: &str, args: OpCompose) -> Result<Self::Composer> {
131        self.inner.compose(ctx, to, args)
132    }
133
134    async fn create_dir(
135        &self,
136        ctx: &OperationContext,
137        path: &str,
138        args: OpCreateDir,
139    ) -> Result<RpCreateDir> {
140        hotpath::measure_async(LABEL_CREATE_DIR, self.inner.create_dir(ctx, path, args)).await
141    }
142
143    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
144        self.inner.read(ctx, path, args).map(HotpathWrapper::new)
145    }
146
147    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
148        self.inner.write(ctx, path, args).map(HotpathWrapper::new)
149    }
150
151    fn copy(
152        &self,
153        ctx: &OperationContext,
154        from: &str,
155        to: &str,
156        args: OpCopy,
157    ) -> Result<Self::Copier> {
158        self.inner
159            .copy(ctx, from, to, args)
160            .map(HotpathWrapper::new)
161    }
162
163    async fn rename(
164        &self,
165        ctx: &OperationContext,
166        from: &str,
167        to: &str,
168        args: OpRename,
169    ) -> Result<RpRename> {
170        hotpath::measure_async(LABEL_RENAME, self.inner.rename(ctx, from, to, args)).await
171    }
172
173    async fn restore(
174        &self,
175        ctx: &OperationContext,
176        path: &str,
177        args: OpRestore,
178    ) -> Result<RpRestore> {
179        hotpath::measure_async(LABEL_RESTORE, self.inner.restore(ctx, path, args)).await
180    }
181
182    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
183        hotpath::measure_async(LABEL_STAT, self.inner.stat(ctx, path, args)).await
184    }
185
186    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
187        self.inner.delete(ctx).map(HotpathWrapper::new)
188    }
189
190    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
191        self.inner.list(ctx, path, args).map(HotpathWrapper::new)
192    }
193
194    async fn presign(
195        &self,
196        ctx: &OperationContext,
197        path: &str,
198        args: OpPresign,
199    ) -> Result<RpPresign> {
200        hotpath::measure_async(LABEL_PRESIGN, self.inner.presign(ctx, path, args)).await
201    }
202}
203
204#[doc(hidden)]
205pub struct HotpathWrapper<R> {
206    inner: R,
207}
208
209impl<R> HotpathWrapper<R> {
210    fn new(inner: R) -> Self {
211        Self { inner }
212    }
213}
214
215impl<R: oio::ReadStream> oio::ReadStream for HotpathWrapper<R> {
216    async fn read(&mut self) -> Result<Buffer> {
217        hotpath::measure_async(LABEL_READER_READ, self.inner.read()).await
218    }
219}
220
221impl<R: oio::Read> oio::Read for HotpathWrapper<R> {
222    async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
223        let (rp, stream) = hotpath::measure_async(LABEL_READ, self.inner.open(range)).await?;
224        Ok((
225            rp,
226            Box::new(HotpathWrapper::new(stream)) as Box<dyn oio::ReadStreamDyn>,
227        ))
228    }
229
230    async fn read(&self, range: BytesRange) -> Result<(RpRead, Buffer)> {
231        hotpath::measure_async(LABEL_READER_READ, self.inner.read(range)).await
232    }
233}
234
235impl<R: oio::Write> oio::Write for HotpathWrapper<R> {
236    async fn write(&mut self, bs: Buffer) -> Result<()> {
237        hotpath::measure_async(LABEL_WRITER_WRITE, self.inner.write(bs)).await
238    }
239
240    async fn copy_from(&mut self, path: &str, args: OpRead, range: BytesRange) -> Result<()> {
241        hotpath::measure_async(LABEL_WRITER_WRITE, self.inner.copy_from(path, args, range)).await
242    }
243
244    async fn close(&mut self) -> Result<Metadata> {
245        hotpath::measure_async(LABEL_WRITER_CLOSE, self.inner.close()).await
246    }
247
248    async fn abort(&mut self) -> Result<()> {
249        hotpath::measure_async(LABEL_WRITER_ABORT, self.inner.abort()).await
250    }
251}
252
253impl<R: oio::List> oio::List for HotpathWrapper<R> {
254    async fn next(&mut self) -> Result<Option<oio::Entry>> {
255        hotpath::measure_async(LABEL_LISTER_NEXT, self.inner.next()).await
256    }
257}
258
259impl<R: oio::Delete> oio::Delete for HotpathWrapper<R> {
260    async fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
261        hotpath::measure_async(LABEL_DELETER_DELETE, self.inner.delete(path, args)).await
262    }
263
264    async fn close(&mut self) -> Result<()> {
265        hotpath::measure_async(LABEL_DELETER_CLOSE, self.inner.close()).await
266    }
267}
268
269impl<C: oio::Copy> oio::Copy for HotpathWrapper<C> {
270    async fn next(&mut self) -> Result<Option<usize>> {
271        hotpath::measure_async(LABEL_COPIER_NEXT, self.inner.next()).await
272    }
273
274    async fn close(&mut self) -> Result<Metadata> {
275        hotpath::measure_async(LABEL_COPIER_CLOSE, self.inner.close()).await
276    }
277
278    async fn abort(&mut self) -> Result<()> {
279        hotpath::measure_async(LABEL_COPIER_ABORT, self.inner.abort()).await
280    }
281}
282
283struct HotpathHttpTransport {
284    inner: HttpTransporter,
285}
286
287impl HttpTransport for HotpathHttpTransport {
288    async fn fetch(&self, req: http::Request<Buffer>) -> Result<http::Response<HttpBody>> {
289        let resp = hotpath::measure_async(LABEL_HTTP_FETCH, self.inner.fetch(req)).await?;
290        let (parts, body) = resp.into_parts();
291        let body = body.map_inner(|stream| Box::new(HotpathStream { inner: stream }));
292        Ok(http::Response::from_parts(parts, body))
293    }
294}
295
296struct HotpathStream<S> {
297    inner: S,
298}
299
300impl<S> Stream for HotpathStream<S>
301where
302    S: Stream<Item = Result<Buffer>> + Unpin + 'static,
303{
304    type Item = Result<Buffer>;
305
306    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
307        let _label = LABEL_HTTP_BODY_POLL;
308        hotpath::measure_block!(_label, self.inner.poll_next_unpin(cx))
309    }
310}
opendal_layer_hotpath/lib.rs

opendal_layer_hotpath/
lib.rs
