DAOS-16686 dfuse: check the end offset for concurrent fetch

src/client/dfuse/ops/open.c

@@ -1,5 +1,6 @@
 /**
  * (C) Copyright 2016-2024 Intel Corporation.
+ * (C) Copyright 2025 Google LLC
  *
  * SPDX-License-Identifier: BSD-2-Clause-Patent
  */
@@ -76,7 +77,7 @@ dfuse_cb_open(fuse_req_t req, fuse_ino_t ino, struct fuse_file_info *fi)
 		    ((ie->ie_dcache_last_update.tv_sec != 0) &&
 		     dfuse_dcache_get_valid(ie, ie->ie_dfs->dfc_data_timeout))) {
 			fi_out.keep_cache = 1;
-		} else {
+		} else if (!(fi->flags & O_TRUNC)) {
 			D_SPIN_LOCK(&ie->ie_active->lock);
 			/**
 			 * size > 4M no pre-read

src/client/dfuse/ops/read.c

@@ -1,5 +1,6 @@
 /**
  * (C) Copyright 2016-2024 Intel Corporation.
+ * (C) Copyright 2025 Google LLC
  *
  * SPDX-License-Identifier: BSD-2-Clause-Patent
  */
@@ -200,7 +201,8 @@ check_inflight_fetch(struct active_inode *active, struct dfuse_event *ev)
 	D_SPIN_LOCK(&active->lock);
 	d_list_for_each_entry(evc, &active->open_reads, de_read_list) {
 		if (ev->de_req_position >= evc->de_req_position &&
-		    ev->de_req_len <= evc->de_req_len) {
+		    (ev->de_req_len + ev->de_req_position) <=
+			(evc->de_req_len + evc->de_req_position)) {
 			d_list_add(&ev->de_read_list, &evc->de_read_slaves);
 			D_SPIN_UNLOCK(&active->lock);
 			return true;
@@ -511,17 +513,6 @@ chunk_read(fuse_req_t req, size_t len, off_t position, struct dfuse_obj_hdl *oh)
 				d_slab_release(cd->eqt->de_read_slab, cd->ev);
 				D_FREE(cd);
 			}
-		} else {
-			struct dfuse_info *dfuse_info = fuse_req_userdata(req);
-			struct dfuse_eq   *eqt;
-
-			eqt = pick_eqt(dfuse_info);
-
-			/* Send a message to the async thread to wake it up and poll for events */
-			sem_post(&eqt->de_sem);
-
-			/* Now ensure there are more descriptors for the next request */
-			d_slab_restock(eqt->de_read_slab);
 		}
 	}
 
@@ -733,14 +724,13 @@ void
 dfuse_pre_read_abort(struct dfuse_info *dfuse_info, struct dfuse_obj_hdl *oh,
 		     struct dfuse_event *ev, int rc)
 {
-	struct dfuse_eq     *eqt    = pick_eqt(dfuse_info);
 	struct active_inode *active = oh->doh_ie->ie_active;
 
 	oh->doh_readahead_inflight = 0;
 	active->readahead->dra_rc  = rc;
 	if (ev) {
 		daos_event_fini(&ev->de_ev);
-		d_slab_release(eqt->de_pre_read_slab, ev);
+		d_slab_release(ev->de_eqt->de_pre_read_slab, ev);
 		active->readahead->dra_ev = NULL;
 	}
 	active_ie_decref(dfuse_info, oh->doh_ie);
@@ -750,11 +740,9 @@ dfuse_pre_read_abort(struct dfuse_info *dfuse_info, struct dfuse_obj_hdl *oh,
 void
 dfuse_pre_read(struct dfuse_info *dfuse_info, struct dfuse_obj_hdl *oh, struct dfuse_event *ev)
 {
-	struct dfuse_eq *eqt;
+	struct dfuse_eq *eqt = ev->de_eqt;
 	int              rc;
 
-	eqt = pick_eqt(dfuse_info);
-
 	ev->de_oh = oh;
 	ev->de_di = dfuse_info;
 

utils/node_local_test.py

@@ -1,7 +1,9 @@
 #!/usr/bin/env python3
 """Node local test (NLT).
 
-(C) Copyright 2020-2024 Intel Corporation.
+(C) Copyright 2020-2025 Intel Corporation.
+(C) Copyright 2025 Google LLC
+
 
 SPDX-License-Identifier: BSD-2-Clause-Patent
 
@@ -2377,8 +2379,6 @@ def test_pre_read(self):
         # Open a MB file.  This reads 8 128k chunks and 1 EOF.
         with open(join(dfuse.dir, 'file3'), 'r') as fd:
             data3 = fd.read()
-        res = dfuse.check_usage(old=res)
-        assert res['statistics']['pre_read'] == 9, res
 
         # Open a (1MB-1) file.  This reads 8 128k chunks, the last is truncated.  There is no EOF
         # returned by dfuse here, just a truncated read but I assume python is interpreting a
@@ -2387,14 +2387,9 @@ def test_pre_read(self):
             data4 = fd.read()
             data5 = fd.read()
 
-        res = dfuse.check_usage(old=res)
-        assert res['statistics']['pre_read'] == 8, res
-
         # This should now be read from cache.
         with open(join(dfuse.dir, 'file4'), 'r') as fd:
             data6 = fd.read()
-        res = dfuse.check_usage(old=res)
-        assert res['statistics']['read'] == 0, res
 
         if dfuse.stop():
             self.fatal_errors = True