// SPDX-License-Identifier: GPL-2.0-only
/* Network filesystem high-level (buffered) writeback.
 *
 * Copyright (C) 2024 Red Hat, Inc. All Rights Reserved.
 * Written by David Howells (dhowells@redhat.com)
 *
 *
 * To support network filesystems with local caching, we manage a situation
 * that can be envisioned like the following:
 *
 *               +---+---+-----+-----+---+----------+
 *    Folios:    |   |   |     |     |   |          |
 *               +---+---+-----+-----+---+----------+
 *
 *                 +------+------+     +----+----+
 *    Upload:      |      |      |.....|    |    |
 *  (Stream 0)     +------+------+     +----+----+
 *
 *               +------+------+------+------+------+
 *    Cache:     |      |      |      |      |      |
 *  (Stream 1)   +------+------+------+------+------+
 *
 * Where we have a sequence of folios of varying sizes that we need to overlay
 * with multiple parallel streams of I/O requests, where the I/O requests in a
 * stream may also be of various sizes (in cifs, for example, the sizes are
 * negotiated with the server; in something like ceph, they may represent the
 * sizes of storage objects).
 *
 * The sequence in each stream may contain gaps and noncontiguous subrequests
 * may be glued together into single vectored write RPCs.
 */

#include <linux/export.h>
#include <linux/fs.h>
#include <linux/mm.h>
#include <linux/pagemap.h>
#include "internal.h"

/*
 * Kill all dirty folios in the event of an unrecoverable error, starting with
 * a locked folio we've already obtained from writeback_iter().
 */
static void netfs_kill_dirty_pages(struct address_space *mapping,
				   struct writeback_control *wbc,
				   struct folio *folio)
{
	int error = 0;

	do {
		enum netfs_folio_trace why = netfs_folio_trace_kill;
		struct netfs_group *group = NULL;
		struct netfs_folio *finfo = NULL;
		void *priv;

		priv = folio_detach_private(folio);
		if (priv) {
			finfo = __netfs_folio_info(priv);
			if (finfo) {
				/* Kill folio from streaming write. */
				group = finfo->netfs_group;
				why = netfs_folio_trace_kill_s;
			} else {
				group = priv;
				if (group == NETFS_FOLIO_COPY_TO_CACHE) {
					/* Kill copy-to-cache folio */
					why = netfs_folio_trace_kill_cc;
					group = NULL;
				} else {
					/* Kill folio with group */
					why = netfs_folio_trace_kill_g;
				}
			}
		}

		trace_netfs_folio(folio, why);

		folio_start_writeback(folio);
		folio_unlock(folio);
		folio_end_writeback(folio);

		netfs_put_group(group);
		kfree(finfo);

	} while ((folio = writeback_iter(mapping, wbc, folio, &error)));
}

/*
 * Create a write request and set it up appropriately for the origin type.
 */
struct netfs_io_request *netfs_create_write_req(struct address_space *mapping,
						struct file *file,
						uoff_t start,
						enum netfs_io_origin origin)
{
	struct netfs_io_request *wreq;
	struct netfs_inode *ictx;
	bool is_cacheable = (origin == NETFS_WRITEBACK ||
			     origin == NETFS_WRITEBACK_SINGLE ||
			     origin == NETFS_PGPRIV2_COPY_TO_CACHE);

	wreq = netfs_alloc_request(mapping, file, start, 0, origin);
	if (IS_ERR(wreq))
		return wreq;

	_enter("R=%x", wreq->debug_id);

	ictx = netfs_inode(wreq->inode);
	if (is_cacheable)
		fscache_begin_write_operation(&wreq->cache_resources, netfs_i_cookie(ictx));
	if (rolling_buffer_init(&wreq->buffer, ITER_SOURCE, wreq->gfp,
				(origin == NETFS_WRITEBACK ||
				 origin == NETFS_WRITEBACK_SINGLE)) < 0)
		goto nomem;

	wreq->cleaned_to = wreq->start;
	if (wreq->cache_resources.dio_size > 1)
		wreq->cache_coll_to = round_down(wreq->start, wreq->cache_resources.dio_size);

	wreq->io_streams[0].stream_nr		= 0;
	wreq->io_streams[0].source		= NETFS_UPLOAD_TO_SERVER;
	wreq->io_streams[0].prepare_write	= ictx->ops->prepare_write;
	wreq->io_streams[0].issue_write		= ictx->ops->issue_write;
	wreq->io_streams[0].collected_to	= start;
	wreq->io_streams[0].transferred		= 0;

	wreq->io_streams[1].stream_nr		= 1;
	wreq->io_streams[1].source		= NETFS_WRITE_TO_CACHE;
	wreq->io_streams[1].collected_to	= start;
	wreq->io_streams[1].transferred		= 0;
	if (fscache_resources_valid(&wreq->cache_resources)) {
		wreq->io_streams[1].avail	= true;
		wreq->io_streams[1].active	= true;
		wreq->io_streams[1].prepare_write = wreq->cache_resources.ops->prepare_write_subreq;
		wreq->io_streams[1].issue_write = wreq->cache_resources.ops->issue_write;
	}

	return wreq;
nomem:
	netfs_put_failed_request(wreq);
	return ERR_PTR(-ENOMEM);
}

/**
 * netfs_prepare_write_failed - Note write preparation failed
 * @subreq: The subrequest to mark
 *
 * Mark a subrequest to note that preparation for write failed.
 */
void netfs_prepare_write_failed(struct netfs_io_subrequest *subreq)
{
	WARN_ON_ONCE(!subreq->error);
	__set_bit(NETFS_SREQ_FAILED, &subreq->flags);
	trace_netfs_sreq(subreq, netfs_sreq_trace_prep_failed);
}
EXPORT_SYMBOL(netfs_prepare_write_failed);

/*
 * Prepare a write subrequest.  We need to allocate a new subrequest
 * if we don't have one.
 */
void netfs_prepare_write(struct netfs_io_request *wreq,
			 struct netfs_io_stream *stream,
			 uoff_t start)
{
	struct netfs_io_subrequest *subreq;
	struct iov_iter *wreq_iter = &wreq->buffer.iter;

	/* Make sure we don't point the iterator at a used-up bvecq struct
	 * being used as a placeholder to prevent the queue from collapsing.
	 * In such a case, extend the queue.
	 */
	if (iov_iter_is_bvecq(wreq_iter) &&
	    !bvecq_acquire_slot(wreq_iter->bvecq, wreq_iter->bvecq_slot))
		rolling_buffer_make_space(&wreq->buffer, wreq->gfp);

	subreq = netfs_alloc_subrequest(wreq, stream->source);
	if (!subreq)
		return;
	subreq->start		= start;
	subreq->stream_nr	= stream->stream_nr;
	subreq->io_iter		= *wreq_iter;

	_enter("R=%x[%x]", wreq->debug_id, subreq->debug_index);

	trace_netfs_sreq(subreq, netfs_sreq_trace_prepare);

	stream->sreq_max_len	= UINT_MAX;
	stream->sreq_max_segs	= INT_MAX;
	switch (stream->source) {
	case NETFS_UPLOAD_TO_SERVER:
		netfs_stat(&netfs_n_wh_upload);
		stream->sreq_max_len = wreq->wsize;
		break;
	case NETFS_WRITE_TO_CACHE:
		netfs_stat(&netfs_n_wh_write);
		break;
	default:
		WARN_ON_ONCE(1);
		break;
	}

	if (stream->prepare_write)
		stream->prepare_write(subreq);

	__set_bit(NETFS_SREQ_IN_PROGRESS, &subreq->flags);

	/* We add to the end of the list whilst the collector may be walking
	 * the list.  The collector only goes nextwards and uses the lock to
	 * remove entries off of the front.
	 */
	spin_lock(&wreq->lock);
	/* Write IN_PROGRESS before pointer to new subreq */
	list_add_tail_release(&subreq->rreq_link, &stream->subrequests);
	if (list_is_first(&subreq->rreq_link, &stream->subrequests)) {
		if (!stream->active) {
			stream->collected_to = subreq->start;
			/* Write list pointers before active flag */
			smp_store_release(&stream->active, true);
		}
	}

	spin_unlock(&wreq->lock);

	stream->construct = subreq;
}

/*
 * Set the I/O iterator for the filesystem/cache to use and dispatch the I/O
 * operation.  The operation may be asynchronous and should call
 * netfs_write_subrequest_terminated() when complete.
 */
static void netfs_do_issue_write(struct netfs_io_stream *stream,
				 struct netfs_io_subrequest *subreq)
{
	struct netfs_io_request *wreq = subreq->rreq;

	_enter("R=%x[%x],%zx", wreq->debug_id, subreq->debug_index, subreq->len);

	if (stream->source == NETFS_WRITE_TO_CACHE &&
	    unlikely(test_bit(NETFS_RREQ_CACHE_STOP, &wreq->flags))) {
		size_t dio_size = wreq->cache_resources.dio_size;
		size_t len, disp;

		disp = subreq->start & (dio_size - 1);
		len = round_up(subreq->len + disp, dio_size);

		subreq->start -= disp;
		subreq->len = len;

		__set_bit(NETFS_SREQ_CANCELLED, &subreq->flags);
		return netfs_write_subrequest_terminated(subreq, subreq->len);
	}

	if (test_bit(NETFS_SREQ_FAILED, &subreq->flags))
		return netfs_write_subrequest_terminated(subreq, subreq->error);

	trace_netfs_sreq(subreq, netfs_sreq_trace_submit);
	stream->issue_write(subreq);
}

void netfs_reissue_write(struct netfs_io_stream *stream,
			 struct netfs_io_subrequest *subreq,
			 struct iov_iter *source)
{
	size_t size = subreq->len - subreq->transferred;

	// TODO: Use encrypted buffer
	subreq->io_iter = *source;
	iov_iter_advance(source, size);
	iov_iter_truncate(&subreq->io_iter, size);

	subreq->retry_count++;
	subreq->error = 0;
	__clear_bit(NETFS_SREQ_MADE_PROGRESS, &subreq->flags);
	__set_bit(NETFS_SREQ_IN_PROGRESS, &subreq->flags);
	netfs_stat(&netfs_n_wh_retry_write_subreq);
	netfs_do_issue_write(stream, subreq);
}

void netfs_issue_write(struct netfs_io_request *wreq,
		       struct netfs_io_stream *stream)
{
	struct netfs_io_subrequest *subreq = stream->construct;

	if (!subreq)
		return;

	stream->construct = NULL;
	subreq->io_iter.count = subreq->len;
	netfs_do_issue_write(stream, subreq);
}

/*
 * Add data to the write subrequest, dispatching each as we fill it up or if it
 * is discontiguous with the previous.  We only fill one part at a time so that
 * we can avoid overrunning the credits obtained (cifs) and try to parallelise
 * content-crypto preparation with network writes.
 */
size_t netfs_advance_write(struct netfs_io_request *wreq,
			   struct netfs_io_stream *stream,
			   uoff_t start, size_t len, bool to_eof)
{
	struct netfs_io_subrequest *subreq = stream->construct;
	size_t part;

	if (!stream->avail) {
		_leave("no write");
		return len;
	}

	_enter("R=%x[%x]", wreq->debug_id, subreq ? subreq->debug_index : 0);

	if (subreq && start != subreq->start + subreq->len) {
		netfs_issue_write(wreq, stream);
		subreq = NULL;
	}

	if (!stream->construct)
		netfs_prepare_write(wreq, stream, start);
	subreq = stream->construct;

	part = umin(stream->sreq_max_len - subreq->len, len);
	_debug("part %zx/%zx %zx/%zx", subreq->len, stream->sreq_max_len, part, len);
	subreq->len += part;
	subreq->nr_segs++;
	stream->submit_extendable_to -= part;

	if (subreq->len >= stream->sreq_max_len ||
	    subreq->nr_segs >= stream->sreq_max_segs ||
	    to_eof) {
		netfs_issue_write(wreq, stream);
		subreq = NULL;
	}

	return part;
}

/*
 * Write some of a pending folio data back to the server.
 */
static int netfs_write_folio(struct netfs_io_request *wreq,
			     struct writeback_control *wbc,
			     struct folio *folio)
{
	struct netfs_io_stream *upload = &wreq->io_streams[0];
	struct netfs_io_stream *cache  = &wreq->io_streams[1];
	struct netfs_io_stream *stream;
	struct netfs_group *fgroup; /* TODO: Use this with ceph */
	struct netfs_folio *finfo;
	size_t iter_off = 0;
	size_t fsize = folio_size(folio), flen = fsize, foff = 0;
	uoff_t fpos = folio_pos(folio), i_size;
	bool to_eof = false, streamw = false;
	bool debug = false;

	_enter("");

	if (rolling_buffer_make_space(&wreq->buffer, wreq->gfp) < 0)
		return -ENOMEM;

	/* netfs_perform_write() may shift i_size around the page or from out
	 * of the page to beyond it, but cannot move i_size into or through the
	 * page since we have it locked.
	 */
	i_size = i_size_read(wreq->inode);

	if (fpos >= i_size) {
		/* mmap beyond eof. */
		_debug("beyond eof");
		folio_start_writeback(folio);
		folio_unlock(folio);
		wreq->nr_group_rel += netfs_folio_written_back(folio);
		netfs_put_group_many(wreq->group, wreq->nr_group_rel);
		wreq->nr_group_rel = 0;
		return 0;
	}

	if (fpos + fsize > wreq->i_size)
		wreq->i_size = i_size;

	fgroup = netfs_folio_group(folio);
	finfo = netfs_folio_info(folio);
	if (finfo) {
		foff = finfo->dirty_offset;
		flen = foff + finfo->dirty_len;
		streamw = true;
	}

	if (flen > i_size - fpos) {
		flen = i_size - fpos;
		if (!streamw)
			folio_zero_segment(folio, flen, fsize);
		to_eof = true;
	} else if (flen == i_size - fpos) {
		to_eof = true;
	}
	flen -= foff;

	_debug("folio %zx %zx %zx", foff, flen, fsize);

	/* Deal with discontinuities in the stream of dirty pages.  These can
	 * arise from a number of sources:
	 *
	 * (1) Intervening non-dirty pages from random-access writes, multiple
	 *     flushers writing back different parts simultaneously and manual
	 *     syncing.
	 *
	 * (2) Partially-written pages from write-streaming.
	 *
	 * (3) Pages that belong to a different write-back group (eg.  Ceph
	 *     snapshots).
	 *
	 * (4) Actually-clean pages that were marked for write to the cache
	 *     when they were read.  Note that these appear as a special
	 *     write-back group.
	 */
	if (fgroup == NETFS_FOLIO_COPY_TO_CACHE) {
		netfs_issue_write(wreq, upload);
	} else if (fgroup != wreq->group) {
		/* We can't write this page to the server yet. */
		kdebug("wrong group");
		folio_redirty_for_writepage(wbc, folio);
		folio_unlock(folio);
		netfs_issue_write(wreq, upload);
		netfs_issue_write(wreq, cache);
		return 0;
	}

	if (foff > 0)
		netfs_issue_write(wreq, upload);
	if (streamw)
		netfs_issue_write(wreq, cache);

	folio_start_writeback(folio);
	folio_unlock(folio);

	if (fgroup == NETFS_FOLIO_COPY_TO_CACHE) {
		if (!cache->avail) {
			trace_netfs_folio(folio, netfs_folio_trace_cancel_copy);
			netfs_issue_write(wreq, upload);
			netfs_folio_written_back(folio);
			return 0;
		}
		trace_netfs_folio(folio, netfs_folio_trace_store_copy);
	} else if (!upload->avail && !cache->avail) {
		trace_netfs_folio(folio, netfs_folio_trace_cancel_store);
		netfs_folio_written_back(folio);
		return 0;
	} else if (!upload->construct) {
		trace_netfs_folio(folio, netfs_folio_trace_store);
	} else {
		trace_netfs_folio(folio, netfs_folio_trace_store_plus);
	}

	/* Attach the folio to the rolling buffer. */
	rolling_buffer_append(&wreq->buffer, folio, wreq->gfp);

	/* Move the submission point forward to allow for write-streaming data
	 * not starting at the front of the page.  We don't do write-streaming
	 * with the cache as the cache requires DIO alignment.
	 *
	 * Also skip uploading for data that's been read and just needs copying
	 * to the cache.
	 */
	for (int s = 0; s < NR_IO_STREAMS; s++) {
		stream = &wreq->io_streams[s];
		stream->submit_off = foff;
		stream->submit_len = flen;
		if (!stream->avail ||
		    (stream->source == NETFS_WRITE_TO_CACHE && streamw) ||
		    (stream->source == NETFS_UPLOAD_TO_SERVER &&
		     fgroup == NETFS_FOLIO_COPY_TO_CACHE)) {
			stream->submit_off = UINT_MAX;
			stream->submit_len = 0;
		}
	}

	/* Attach the folio to one or more subrequests.  For a big folio, we
	 * could end up with thousands of subrequests if the wsize is small -
	 * but we might need to wait during the creation of subrequests for
	 * network resources (eg. SMB credits).
	 */
	for (;;) {
		ssize_t part;
		size_t lowest_off = ULONG_MAX;
		int choose_s = -1;

		/* Always add to the lowest-submitted stream first. */
		for (int s = 0; s < NR_IO_STREAMS; s++) {
			stream = &wreq->io_streams[s];
			if (stream->submit_len > 0 &&
			    stream->submit_off < lowest_off) {
				lowest_off = stream->submit_off;
				choose_s = s;
			}
		}

		if (choose_s < 0)
			break;
		stream = &wreq->io_streams[choose_s];

		/* Advance the iterator(s). */
		if (stream->submit_off > iter_off) {
			rolling_buffer_advance(&wreq->buffer, stream->submit_off - iter_off);
			iter_off = stream->submit_off;
		}

		atomic64_set(&wreq->issued_to, fpos + stream->submit_off);
		stream->submit_extendable_to = fsize - stream->submit_off;
		part = netfs_advance_write(wreq, stream, fpos + stream->submit_off,
					   stream->submit_len, to_eof);
		stream->submit_off += part;
		if (part > stream->submit_len)
			stream->submit_len = 0;
		else
			stream->submit_len -= part;
		if (part > 0)
			debug = true;
	}

	if (fsize > iter_off)
		rolling_buffer_advance(&wreq->buffer, fsize - iter_off);
	atomic64_set(&wreq->issued_to, fpos + fsize);

	if (!debug)
		kdebug("R=%x: No submit", wreq->debug_id);

	if (foff + flen < fsize)
		for (int s = 0; s < NR_IO_STREAMS; s++)
			netfs_issue_write(wreq, &wreq->io_streams[s]);

	_leave(" = 0");
	return 0;
}

/*
 * End the issuing of writes, letting the collector know we're done.
 */
static void netfs_end_issue_write(struct netfs_io_request *wreq)
{
	bool needs_poke = true;

	netfs_all_subreqs_queued(wreq);

	for (int s = 0; s < NR_IO_STREAMS; s++) {
		struct netfs_io_stream *stream = &wreq->io_streams[s];

		if (!stream->active)
			continue;
		if (!list_empty(&stream->subrequests))
			needs_poke = false;
		netfs_issue_write(wreq, stream);
	}

	if (needs_poke)
		netfs_wake_collector(wreq);
}

/*
 * Write some of the pending data back to the server
 */
int netfs_writepages(struct address_space *mapping,
		     struct writeback_control *wbc)
{
	struct netfs_inode *ictx = netfs_inode(mapping->host);
	struct netfs_io_request *wreq = NULL;
	struct folio *folio;
	int error = 0;

	if (!netfs_wb_begin(ictx, wbc->sync_mode == WB_SYNC_NONE))
		return 0;

	/* Need the first folio to be able to set up the op. */
	folio = writeback_iter(mapping, wbc, NULL, &error);
	if (!folio)
		goto out;

	wreq = netfs_create_write_req(mapping, NULL, folio_pos(folio), NETFS_WRITEBACK);
	if (IS_ERR(wreq)) {
		error = PTR_ERR(wreq);
		goto couldnt_start;
	}

	__set_bit(NETFS_RREQ_OFFLOAD_COLLECTION, &wreq->flags);
	trace_netfs_write(wreq, netfs_write_trace_writeback);
	netfs_stat(&netfs_n_wh_writepages);

	do {
		_debug("wbiter %lx %llx", folio->index, atomic64_read(&wreq->issued_to));

		/* It appears we don't have to handle cyclic writeback wrapping. */
		WARN_ON_ONCE(wreq && folio_pos(folio) < atomic64_read(&wreq->issued_to));

		if (netfs_folio_group(folio) != NETFS_FOLIO_COPY_TO_CACHE &&
		    unlikely(!test_bit(NETFS_RREQ_UPLOAD_TO_SERVER, &wreq->flags))) {
			set_bit(NETFS_RREQ_UPLOAD_TO_SERVER, &wreq->flags);
			wreq->netfs_ops->begin_writeback(wreq);
		}

		error = netfs_write_folio(wreq, wbc, folio);
		if (error == -ENOMEM) {
			folio_redirty_for_writepage(wbc, folio);
			folio_unlock(folio);
		}
	} while ((folio = writeback_iter(mapping, wbc, folio, &error)));

	netfs_end_issue_write(wreq);
	netfs_wake_collector(wreq);

	netfs_put_request(wreq, netfs_rreq_trace_put_return);
	_leave(" = %d", error);
	return error;

couldnt_start:
	if (error == -ENOMEM) {
		folio_redirty_for_writepage(wbc, folio);
		folio_unlock(folio);
		folio = writeback_iter(mapping, wbc, folio, &error);
		WARN_ON_ONCE(folio != NULL);
	} else {
		netfs_kill_dirty_pages(mapping, wbc, folio);
	}
out:
	netfs_wb_end(ictx);
	_leave(" = %d", error);
	return error;
}
EXPORT_SYMBOL(netfs_writepages);

/**
 * netfs_writeback_single - Write back a monolithic payload
 * @mapping: The mapping to write from
 * @wbc: Hints from the VM
 * @iter: Data to write.
 *
 * Write a monolithic, non-pagecache object back to the server and/or the
 * cache.  The caller must explicitly set NETFS_RREQ_UPLOAD_TO_SERVER when
 * initialising the request if it wants the data to be written to the server
 * (for AFS directories and symlinks, this is not possible; things like mkdir,
 * symlink, rmdir and unlink must be used instead).
 *
 * Return: 0 if successful; 1 if skipped due to lock conflict and WB_SYNC_NONE;
 * or a negative error code.
 */
int netfs_writeback_single(struct address_space *mapping,
			   struct writeback_control *wbc,
			   struct iov_iter *iter)
{
	struct netfs_io_request *wreq;
	struct netfs_inode *ictx = netfs_inode(mapping->host);
	int ret;

	if (!netfs_wb_begin(ictx, wbc->sync_mode == WB_SYNC_NONE)) {
		/* The VFS will have undirtied the inode. */
		netfs_single_mark_inode_dirty(&ictx->inode);
		return 1;
	}

	wreq = netfs_create_write_req(mapping, NULL, 0, NETFS_WRITEBACK_SINGLE);
	if (IS_ERR(wreq)) {
		ret = PTR_ERR(wreq);
		goto couldnt_start;
	}

	wreq->buffer.iter = *iter;
	wreq->len = iov_iter_count(iter);
	wreq->submitted = wreq->len;

	__set_bit(NETFS_RREQ_OFFLOAD_COLLECTION, &wreq->flags);
	trace_netfs_write(wreq, netfs_write_trace_writeback_single);
	netfs_stat(&netfs_n_wh_writepages);

	/* Start writing to the server only if asked. */
	if (test_bit(NETFS_RREQ_UPLOAD_TO_SERVER, &wreq->flags))
		wreq->netfs_ops->begin_writeback(wreq);

	for (int s = 0; s < NR_IO_STREAMS; s++) {
		struct netfs_io_subrequest *subreq;
		struct netfs_io_stream *stream = &wreq->io_streams[s];

		if (!stream->avail)
			continue;

		netfs_prepare_write(wreq, stream, 0);

		subreq = stream->construct;
		subreq->len = wreq->len;
		stream->submit_len = subreq->len;
		stream->submit_extendable_to = round_up(wreq->len, PAGE_SIZE);

		netfs_issue_write(wreq, stream);
	}

	netfs_all_subreqs_queued(wreq);
	netfs_wake_collector(wreq);

	/* TODO: Might want to be async here if WB_SYNC_NONE, but then need to
	 * wait before modifying.
	 */
	ret = netfs_wait_for_write(wreq);
	if (ret > 0)
		ret = 0;

	netfs_put_request(wreq, netfs_rreq_trace_put_return);
	_leave(" = %d", ret);
	return ret;

couldnt_start:
	netfs_wb_end(ictx);
	_leave(" = %d", ret);
	return ret;
}
EXPORT_SYMBOL(netfs_writeback_single);