refactor(ext/cache): Remove custom shutdown and use fast async ops (d…

…enoland#20107)

The original implementation of `Cache` used a custom `shutdown` method
on the resource, but to simplify fast streams work we're going to move
this to an op of its own.

While we're in here, we're going to replace `opAsync` with
`ensureFastOps`. `op2` work will have to wait because of some
limitations to our async support, however.

ext/cache/01_cache.js

@@ -1,5 +1,5 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
-
+// deno-lint-ignore-file camelcase
 const core = globalThis.Deno.core;
 import * as webidl from "ext:deno_webidl/00_webidl.js";
 const primordials = globalThis.__bootstrap.primordials;
@@ -20,7 +20,15 @@ import { toInnerResponse } from "ext:deno_fetch/23_response.js";
 import { URLPrototype } from "ext:deno_url/00_url.js";
 import { getHeader } from "ext:deno_fetch/20_headers.js";
 import { readableStreamForRid } from "ext:deno_web/06_streams.js";
-
+const {
+ op_cache_delete,
+ op_cache_match,
+ op_cache_put,
+ op_cache_put_finish,
+ op_cache_storage_delete,
+ op_cache_storage_has,
+ op_cache_storage_open,
+} = core.ensureFastOps();
 class CacheStorage {
  constructor() {
  webidl.illegalConstructor();
@@ -31,7 +39,7 @@ class CacheStorage {
  const prefix = "Failed to execute 'open' on 'CacheStorage'";
  webidl.requiredArguments(arguments.length, 1, prefix);
  cacheName = webidl.converters["DOMString"](cacheName, prefix, "Argument 1");
- const cacheId = await core.opAsync("op_cache_storage_open", cacheName);
+ const cacheId = await op_cache_storage_open(cacheName);
  const cache = webidl.createBranded(Cache);
  cache[_id] = cacheId;
  return cache;
@@ -42,15 +50,15 @@ class CacheStorage {
  const prefix = "Failed to execute 'has' on 'CacheStorage'";
  webidl.requiredArguments(arguments.length, 1, prefix);
  cacheName = webidl.converters["DOMString"](cacheName, prefix, "Argument 1");
- return await core.opAsync("op_cache_storage_has", cacheName);
+ return await op_cache_storage_has(cacheName);
  }
 
  async delete(cacheName) {
  webidl.assertBranded(this, CacheStoragePrototype);
  const prefix = "Failed to execute 'delete' on 'CacheStorage'";
  webidl.requiredArguments(arguments.length, 1, prefix);
  cacheName = webidl.converters["DOMString"](cacheName, prefix, "Argument 1");
- return await core.opAsync("op_cache_storage_delete", cacheName);
+ return await op_cache_storage_delete(cacheName);
  }
 }
 
@@ -124,8 +132,7 @@ class Cache {
  reqUrl.hash = "";
 
  // Step 9-11.
- const rid = await core.opAsync(
- "op_cache_put",
+ const rid = await op_cache_put(
  {
  cacheId: this[_id],
  // deno-lint-ignore prefer-primordials
@@ -142,7 +149,7 @@ class Cache {
  while (true) {
  const { value, done } = await reader.read();
  if (done) {
- await core.shutdown(rid);
+ await op_cache_put_finish(rid);
  break;
  }
  await core.writeAll(rid, value);
@@ -196,7 +203,7 @@ class Cache {
  ) {
  r = new Request(request);
  }
- return await core.opAsync("op_cache_delete", {
+ return await op_cache_delete({
  cacheId: this[_id],
  requestUrl: r.url,
  });
@@ -240,8 +247,7 @@ class Cache {
  const url = new URL(r.url);
  url.hash = "";
  const innerRequest = toInnerRequest(r);
- const matchResult = await core.opAsync(
- "op_cache_match",
+ const matchResult = await op_cache_match(
  {
  cacheId: this[_id],
  // deno-lint-ignore prefer-primordials

ext/cache/lib.rs

@@ -14,6 +14,7 @@ use deno_core::ByteString;
 use deno_core::OpState;
 use deno_core::Resource;
 use deno_core::ResourceId;
+
 mod sqlite;
 pub use sqlite::SqliteBackedCache;
 
@@ -28,6 +29,7 @@ deno_core::extension!(deno_cache,
  op_cache_storage_has<CA>,
  op_cache_storage_delete<CA>,
  op_cache_put<CA>,
+ op_cache_put_finish<CA>,
  op_cache_match<CA>,
  op_cache_delete<CA>,
  ],
@@ -86,16 +88,24 @@ pub struct CacheDeleteRequest {
  pub request_url: String,
 }
 
-#[async_trait]
-pub trait Cache: Clone {
+#[async_trait(?Send)]
+pub trait Cache: Clone + 'static {
+ type CachePutResourceType: Resource;
+
  async fn storage_open(&self, cache_name: String) -> Result<i64, AnyError>;
  async fn storage_has(&self, cache_name: String) -> Result<bool, AnyError>;
  async fn storage_delete(&self, cache_name: String) -> Result<bool, AnyError>;
 
- async fn put(
+ /// Create a put request.
+ async fn put_create(
  &self,
  request_response: CachePutRequest,
- ) -> Result<Option<Rc<dyn Resource>>, AnyError>;
+ ) -> Result<Option<Rc<Self::CachePutResourceType>>, AnyError>;
+ /// Complete a put request.
+ async fn put_finish(
+ &self,
+ resource: Rc<Self::CachePutResourceType>,
+ ) -> Result<(), AnyError>;
  async fn r#match(
  &self,
  request: CacheMatchRequest,
@@ -113,7 +123,7 @@ pub async fn op_cache_storage_open<CA>(
  cache_name: String,
 ) -> Result<i64, AnyError>
 where
- CA: Cache + 'static,
+ CA: Cache,
 {
  let cache = get_cache::<CA>(&state)?;
  cache.storage_open(cache_name).await
@@ -125,7 +135,7 @@ pub async fn op_cache_storage_has<CA>(
  cache_name: String,
 ) -> Result<bool, AnyError>
 where
- CA: Cache + 'static,
+ CA: Cache,
 {
  let cache = get_cache::<CA>(&state)?;
  cache.storage_has(cache_name).await
@@ -137,7 +147,7 @@ pub async fn op_cache_storage_delete<CA>(
  cache_name: String,
 ) -> Result<bool, AnyError>
 where
- CA: Cache + 'static,
+ CA: Cache,
 {
  let cache = get_cache::<CA>(&state)?;
  cache.storage_delete(cache_name).await
@@ -149,10 +159,10 @@ pub async fn op_cache_put<CA>(
  request_response: CachePutRequest,
 ) -> Result<Option<ResourceId>, AnyError>
 where
- CA: Cache + 'static,
+ CA: Cache,
 {
  let cache = get_cache::<CA>(&state)?;
- match cache.put(request_response).await? {
+ match cache.put_create(request_response).await? {
  Some(resource) => {
  let rid = state.borrow_mut().resource_table.add_rc_dyn(resource);
  Ok(Some(rid))
@@ -161,13 +171,29 @@ where
  }
 }
 
+#[op]
+pub async fn op_cache_put_finish<CA>(
+ state: Rc<RefCell<OpState>>,
+ rid: ResourceId,
+) -> Result<(), AnyError>
+where
+ CA: Cache,
+{
+ let cache = get_cache::<CA>(&state)?;
+ let resource = state
+ .borrow_mut()
+ .resource_table
+ .get::<CA::CachePutResourceType>(rid)?;
+ cache.put_finish(resource).await
+}
+
 #[op]
 pub async fn op_cache_match<CA>(
  state: Rc<RefCell<OpState>>,
  request: CacheMatchRequest,
 ) -> Result<Option<CacheMatchResponse>, AnyError>
 where
- CA: Cache + 'static,
+ CA: Cache,
 {
  let cache = get_cache::<CA>(&state)?;
  match cache.r#match(request).await? {
@@ -186,15 +212,15 @@ pub async fn op_cache_delete<CA>(
  request: CacheDeleteRequest,
 ) -> Result<bool, AnyError>
 where
- CA: Cache + 'static,
+ CA: Cache,
 {
  let cache = get_cache::<CA>(&state)?;
  cache.delete(request).await
 }
 
 pub fn get_cache<CA>(state: &Rc<RefCell<OpState>>) -> Result<CA, AnyError>
 where
- CA: Cache + 'static,
+ CA: Cache,
 {
  let mut state = state.borrow_mut();
  if let Some(cache) = state.try_borrow::<CA>() {

ext/cache/sqlite.rs

@@ -92,8 +92,10 @@ impl SqliteBackedCache {
  }
 }
 
-#[async_trait]
+#[async_trait(?Send)]
 impl Cache for SqliteBackedCache {
+ type CachePutResourceType = CachePutResource;
+
  /// Open a cache storage. Internally, this creates a row in the
  /// sqlite db if the cache doesn't exist and returns the internal id
  /// of the cache.
@@ -167,10 +169,10 @@ impl Cache for SqliteBackedCache {
  .await?
  }
 
- async fn put(
+ async fn put_create(
  &self,
  request_response: CachePutRequest,
- ) -> Result<Option<Rc<dyn Resource>>, AnyError> {
+ ) -> Result<Option<Rc<CachePutResource>>, AnyError> {
  let db = self.connection.clone();
  let cache_storage_dir = self.cache_storage_dir.clone();
  let now = SystemTime::now().duration_since(UNIX_EPOCH)?;
@@ -202,6 +204,13 @@ impl Cache for SqliteBackedCache {
  }
  }
 
+ async fn put_finish(
+ &self,
+ resource: Rc<CachePutResource>,
+ ) -> Result<(), AnyError> {
+ resource.write_to_cache().await
+ }
+
  async fn r#match(
  &self,
  request: CacheMatchRequest,
@@ -346,7 +355,7 @@ impl CachePutResource {
  Ok(data.len())
  }
 
- async fn shutdown(self: Rc<Self>) -> Result<(), AnyError> {
+ async fn write_to_cache(self: Rc<Self>) -> Result<(), AnyError> {
  let resource = deno_core::RcRef::map(&self, |r| &r.file);
  let mut file = resource.borrow_mut().await;
  file.flush().await?;
@@ -377,10 +386,6 @@ impl Resource for CachePutResource {
  }
 
  deno_core::impl_writable!();
-
- fn shutdown(self: Rc<Self>) -> AsyncResult<()> {
- Box::pin(self.shutdown())
- }
 }
 
 pub struct CacheResponseResource {