DAOS-13216 dfuse: Add a pre-read feature for non-cached files.

src/client/dfuse/dfuse.h

@@ -92,12 +92,49 @@ struct dfuse_readdir_entry {
 	off_t dre_next_offset;
 };
 
+/* Preread.
+ *
+ * If a file is opened when caching is on but the file is not cached and the size small enough
+ * to fit in a single buffer then start a pre-read of the file on open, then when reads do occur
+ * they can happen directly from the buffer.  For 'linear reads' of a file this means that the
+ * read can be triggered sooner and performed as one dfs request.  Make use of the pre-read
+ * code to only use this for trivial reads, if a file is not read linearly or it's written to
+ * then back off to the regular behavior, which will likely use the kernel cache.
+ *
+ * This works by creating a new descriptor which is pointed to by the open handle, on open dfuse
+ * decides if it will use pre-read and if so allocate a new descriptor, add it to the open handle
+ * and then once it's replied to the open immediately issue a read.  The new descriptor includes
+ * a lock which is locked by open before it replies to the kernel request and unlocked by the dfs
+ * read callback.  Read requests then take the lock to ensure the dfs read is complete and reply
+ * directly with the data in the buffer.
+ *
+ * This works up to the buffer size minus one, the pre-read tries to read the expected file size
+ * plus one so if the file size has changed then dfuse will detect this and back off to regular
+ * read.
+ *
+ * A dfuse_event is hung off this new descriptor and these come from the same pool as regular reads,
+ * this buffer is kept as long as it's needed but released as soon as possible, either on error or
+ * when EOF is returned to the kernel.  If it's still present on release then it's freed then.
+ *
+ * TODO: The current issue with pread is that DFuse keeps a single cache time per inode and this
+ * is used for both data and metadata.  For this feature to work the inode cache needs to be valid
+ * and the data cache needs to be empty.
+ */
+struct dfuse_read_ahead {
+	pthread_mutex_t     dra_lock;
+	struct dfuse_event *dra_ev;
+	int                 dra_rc;
+};
+
 /** what is returned as the handle for fuse fuse_file_info on create/open/opendir */
 struct dfuse_obj_hdl {
 	/** pointer to dfs_t */
 	dfs_t                    *doh_dfs;
 	/** the DFS object handle.  Not created for directories. */
 	dfs_obj_t                *doh_obj;
+
+	struct dfuse_read_ahead  *doh_readahead;
+
 	/** the inode entry for the file */
 	struct dfuse_inode_entry *doh_ie;
 
@@ -213,7 +250,10 @@ struct dfuse_event {
 	struct dfuse_eq              *de_eqt;
 	struct dfuse_obj_hdl         *de_oh;
 	off_t                         de_req_position; /**< The file position requested by fuse */
-	size_t                        de_req_len;
+	union {
+		size_t de_req_len;
+		size_t de_readahead_len;
+	};
 	void (*de_complete_cb)(struct dfuse_event *ev);
 };
 
@@ -543,7 +583,7 @@ struct fuse_lowlevel_ops dfuse_ops;
 #define DFUSE_REPLY_ENTRY(inode, req, entry)                                                       \
 	do {                                                                                       \
 		int __rc;                                                                          \
-		DFUSE_TRA_DEBUG(inode, "Returning entry inode %#lx mode %#o size %zi",             \
+		DFUSE_TRA_DEBUG(inode, "Returning entry inode %#lx mode %#o size %#zx",            \
 				(entry).attr.st_ino, (entry).attr.st_mode, (entry).attr.st_size);  \
 		if (entry.attr_timeout > 0) {                                                      \
 			(inode)->ie_stat = entry.attr;                                             \
@@ -695,6 +735,9 @@ dfuse_cache_evict(struct dfuse_inode_entry *ie);
 bool
 dfuse_cache_get_valid(struct dfuse_inode_entry *ie, double max_age, double *timeout);
 
+void
+dfuse_pre_read(struct dfuse_projection_info *fs_handle, struct dfuse_obj_hdl *oh);
+
 int
 check_for_uns_ep(struct dfuse_projection_info *fs_handle,
 		 struct dfuse_inode_entry *ie, char *attr, daos_size_t len);

src/client/dfuse/ops/open.c

@@ -16,6 +16,7 @@ dfuse_cb_open(fuse_req_t req, fuse_ino_t ino, struct fuse_file_info *fi)
 	struct dfuse_obj_hdl         *oh     = NULL;
 	struct fuse_file_info         fi_out = {0};
 	int                           rc;
+	bool                          prefetch = false;
 
 	rlink = d_hash_rec_find(&fs_handle->dpi_iet, &ino, sizeof(ino));
 	if (!rlink) {
@@ -58,6 +59,7 @@ dfuse_cb_open(fuse_req_t req, fuse_ino_t ino, struct fuse_file_info *fi)
 			fi_out.keep_cache = 1;
 		} else if (dfuse_cache_get_valid(ie, ie->ie_dfs->dfc_data_timeout, NULL)) {
 			fi_out.keep_cache = 1;
+			prefetch          = true;
 		}
 
 		if (fi_out.keep_cache)
@@ -89,9 +91,25 @@ dfuse_cb_open(fuse_req_t req, fuse_ino_t ino, struct fuse_file_info *fi)
 
 	atomic_fetch_add_relaxed(&ie->ie_open_count, 1);
 
+	/* Enable this for smaller files, up to but not including the max read size.  This means
+	 * that files of 1MiB or larger will not be pre-read but smaller files will.  One extra
+	 * unused byte at the end of the buffer will be used to detect if the file has grown
+	 * since dfuse last observed the size.
+	 */
+	if (prefetch && ie->ie_stat.st_size > 0 && ie->ie_stat.st_size < DFUSE_MAX_READ) {
+		D_ALLOC_PTR(oh->doh_readahead);
+		if (oh->doh_readahead) {
+			D_MUTEX_INIT(&oh->doh_readahead->dra_lock, 0);
+			D_MUTEX_LOCK(&oh->doh_readahead->dra_lock);
+		}
+	}
+
 	d_hash_rec_decref(&fs_handle->dpi_iet, rlink);
 	DFUSE_REPLY_OPEN(oh, req, &fi_out);
 
+	if (oh->doh_readahead)
+		dfuse_pre_read(fs_handle, oh);
+
 	return;
 err:
 	d_hash_rec_decref(&fs_handle->dpi_iet, rlink);
@@ -112,6 +130,26 @@ dfuse_cb_release(fuse_req_t req, fuse_ino_t ino, struct fuse_file_info *fi)
 
 	DFUSE_TRA_DEBUG(oh, "Closing %d %d", oh->doh_caching, oh->doh_keep_cache);
 
+	if (oh->doh_readahead) {
+		struct dfuse_event *ev;
+
+		/* Grab this lock first to ensure that the read cb has been completed.  The
+		 * callback might register an error and release ev so do not read it's value
+		 * until after this has completed.
+		 */
+		D_MUTEX_LOCK(&oh->doh_readahead->dra_lock);
+		D_MUTEX_UNLOCK(&oh->doh_readahead->dra_lock);
+
+		ev = oh->doh_readahead->dra_ev;
+
+		D_MUTEX_DESTROY(&oh->doh_readahead->dra_lock);
+		if (ev) {
+			daos_event_fini(&ev->de_ev);
+			d_slab_release(ev->de_eqt->de_read_slab, ev);
+		}
+		D_FREE(oh->doh_readahead);
+	}
+
 	/* If the file was read from then set the cache time for future use, however if the
 	 * file was written to then evict the cache.
 	 * The problem here is that if the file was written to then the contents will be in the

src/client/dfuse/ops/read.c

@@ -30,7 +30,7 @@ dfuse_cb_read_complete(struct dfuse_event *ev)
 
 	if (ev->de_len == 0) {
 		DFUSE_TRA_DEBUG(oh, "%#zx-%#zx requested (EOF)", ev->de_req_position,
-				ev->de_req_position + ev->de_iov.iov_buf_len - 1);
+				ev->de_req_position + ev->de_req_len - 1);
 
 		DFUSE_REPLY_BUFQ(oh, ev->de_req, ev->de_iov.iov_buf, ev->de_len);
 		D_GOTO(release, 0);
@@ -51,6 +51,55 @@ dfuse_cb_read_complete(struct dfuse_event *ev)
 	d_slab_release(ev->de_eqt->de_read_slab, ev);
 }
 
+static bool
+dfuse_readahead_reply(fuse_req_t req, size_t len, off_t position, struct dfuse_obj_hdl *oh)
+{
+	size_t reply_len;
+
+	if (oh->doh_readahead->dra_rc) {
+		DFUSE_REPLY_ERR_RAW(oh, req, oh->doh_readahead->dra_rc);
+		return true;
+	}
+
+	if (!oh->doh_linear_read || oh->doh_readahead->dra_ev == NULL) {
+		DFUSE_TRA_DEBUG(oh, "Readahead disabled");
+		return false;
+	}
+
+	if (oh->doh_linear_read_pos != position) {
+		DFUSE_TRA_DEBUG(oh, "disabling readahead");
+		daos_event_fini(&oh->doh_readahead->dra_ev);
+		d_slab_release(oh->doh_readahead->dra_ev->de_eqt->de_read_slab,
+			       oh->doh_readahead->dra_ev);
+		oh->doh_readahead->dra_ev = NULL;
+		return false;
+	}
+
+	oh->doh_linear_read_pos = position + len;
+	if (position + len >= oh->doh_readahead->dra_ev->de_readahead_len) {
+		oh->doh_linear_read_eof = true;
+	}
+
+	/* At this point there is a buffer of known length that contains the data, and a read
+	 * request.
+	 * If the attempted read is bigger than the data then it will be truncated.
+	 * It the atttempted read is smaller than the buffer it will be met in full.
+	 */
+
+	if (position + len < oh->doh_readahead->dra_ev->de_readahead_len) {
+		reply_len = len;
+		DFUSE_TRA_DEBUG(oh, "%#zx-%#zx read", position, position + reply_len - 1);
+	} else {
+		/* The read will be truncated */
+		reply_len = oh->doh_readahead->dra_ev->de_readahead_len - position;
+		DFUSE_TRA_DEBUG(oh, "%#zx-%#zx read %#zx-%#zx not read (truncated)", position,
+				position + reply_len - 1, position + reply_len, position + len - 1);
+	}
+
+	DFUSE_REPLY_BUFQ(oh, req, oh->doh_readahead->dra_ev->de_iov.iov_buf + position, reply_len);
+	return true;
+}
+
 void
 dfuse_cb_read(fuse_req_t req, fuse_ino_t ino, size_t len, off_t position, struct fuse_file_info *fi)
 {
@@ -68,10 +117,36 @@ dfuse_cb_read(fuse_req_t req, fuse_ino_t ino, size_t len, off_t position, struct
 		DFUSE_TRA_DEBUG(oh, "Returning EOF early without round trip %#zx", position);
 		oh->doh_linear_read_eof = false;
 		oh->doh_linear_read     = false;
+
+		if (oh->doh_readahead) {
+			ev = oh->doh_readahead->dra_ev;
+
+			D_MUTEX_LOCK(&oh->doh_readahead->dra_lock);
+			ev                        = oh->doh_readahead->dra_ev;
+			oh->doh_readahead->dra_ev = NULL;
+			D_MUTEX_UNLOCK(&oh->doh_readahead->dra_lock);
+
+			if (ev) {
+				daos_event_fini(&ev->de_ev);
+				d_slab_release(ev->de_eqt->de_read_slab, ev);
+			}
+		}
 		DFUSE_REPLY_BUFQ(oh, req, NULL, 0);
 		return;
 	}
 
+	if (oh->doh_readahead) {
+		bool replied;
+
+		D_MUTEX_LOCK(&oh->doh_readahead->dra_lock);
+		replied = dfuse_readahead_reply(req, len, position, oh);
+		D_MUTEX_UNLOCK(&oh->doh_readahead->dra_lock);
+
+		if (replied) {
+			return;
+		}
+	}
+
 	eqt = &fs_handle->dpi_eqt[eqt_idx % fs_handle->dpi_eqt_count];
 
 	ev = d_slab_acquire(eqt->de_read_slab);
@@ -131,3 +206,83 @@ dfuse_cb_read(fuse_req_t req, fuse_ino_t ino, size_t len, off_t position, struct
 		d_slab_release(eqt->de_read_slab, ev);
 	}
 }
+
+static void
+dfuse_cb_pre_read_complete(struct dfuse_event *ev)
+{
+	struct dfuse_obj_hdl *oh = ev->de_oh;
+
+	oh->doh_readahead->dra_rc = ev->de_ev.ev_error;
+
+	if (ev->de_ev.ev_error != 0) {
+		oh->doh_readahead->dra_rc = ev->de_ev.ev_error;
+		daos_event_fini(&ev->de_ev);
+		d_slab_release(ev->de_eqt->de_read_slab, ev);
+		oh->doh_readahead->dra_ev = NULL;
+	}
+
+	/* If the length is not as expected then the stat size was incorrect so do not use this
+	 * read and fallback into the normal read case.
+	 */
+	if (ev->de_len != ev->de_readahead_len) {
+		daos_event_fini(&ev->de_ev);
+		d_slab_release(ev->de_eqt->de_read_slab, ev);
+		oh->doh_readahead->dra_ev = NULL;
+	}
+
+	D_MUTEX_UNLOCK(&oh->doh_readahead->dra_lock);
+}
+
+void
+dfuse_pre_read(struct dfuse_projection_info *fs_handle, struct dfuse_obj_hdl *oh)
+{
+	struct dfuse_eq    *eqt;
+	int                 rc;
+	struct dfuse_event *ev;
+	uint64_t            eqt_idx;
+	size_t              len = oh->doh_ie->ie_stat.st_size;
+
+	eqt_idx = atomic_fetch_add_relaxed(&fs_handle->dpi_eqt_idx, 1);
+
+	eqt = &fs_handle->dpi_eqt[eqt_idx % fs_handle->dpi_eqt_count];
+
+	ev = d_slab_acquire(eqt->de_read_slab);
+	if (ev == NULL)
+		D_GOTO(err, rc = ENOMEM);
+
+	/* Request a read one byte bigger than the expected file size, this way we should see
+	 * a truncated read and through that will be able to detect if the file has been
+	 * modified
+	 */
+	ev->de_iov.iov_len   = len + 1;
+	ev->de_req           = 0;
+	ev->de_sgl.sg_nr     = 1;
+	ev->de_oh            = oh;
+	ev->de_readahead_len = len;
+	ev->de_req_position  = 0;
+
+	ev->de_complete_cb        = dfuse_cb_pre_read_complete;
+	oh->doh_readahead->dra_ev = ev;
+
+	rc = dfs_read(oh->doh_dfs, oh->doh_obj, &ev->de_sgl, 0, &ev->de_len, &ev->de_ev);
+	if (rc != 0) {
+		D_GOTO(err, rc);
+		return;
+	}
+
+	/* 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);
+
+	return;
+err:
+	oh->doh_readahead->dra_rc = rc;
+	if (ev) {
+		daos_event_fini(&ev->de_ev);
+		d_slab_release(eqt->de_read_slab, ev);
+		oh->doh_readahead->dra_ev = NULL;
+	}
+	D_MUTEX_UNLOCK(&oh->doh_readahead->dra_lock);
+}