iov_iter 구조체에 대해 한 번 정리해 보고자 한다.

iov_iter를 본격적으로 설명해 보기 전에, 먼저 readv()에 대해서 설명할 필요가 있다.
writev()도 물론 가능하지만, 편의상 readv()에 대해서만 알아보도록 하자.

readv() syscall

readv() syscall의 prototype을 보면 다음과 같다.

struct iovec {
           void   *iov_base;  /* Starting address */
           size_t  iov_len;   /* Size of the memory pointed to by iov_base. */
};

ssize_t readv(int fd, const struct iovec *iov, int iovcnt);

여러 버퍼들을 하나의 iovec의 배열을 통해 묶어서 readv() syscall 호출을 통해 읽어올 수 있다.

필요성에 대해 생각해 보면, 만약 산재되어 있는 버퍼의 양이 굉장히 많은 경우, 각각에 대해 read()를 호출하게 된다면 context switch에 의한 overhead가 상당하게 된다.
반면에 readv()의 경우 한 번의 syscall로 여러 번의 read()를 한 효과를 낼 수 있게 되어 이러한 overhead를 줄일 수 있는 장점이 있다.

이런 식으로 userland에서는 iovec을 통해 커널에 넘겨주게 되고, 커널이 받은 iovec을 관리하게 된다. 그리고 이 iovec을 관리하는 데에 사용되는 구조체가 바로 iov_iter이다. (물론 iovec 외에 다른 구조체들에도 쓰인다.)

iov_iter에 대해 알아보자.

struct iov_iter

#define ITER_SOURCE	1	// == WRITE
#define ITER_DEST	0	// == READ

struct iov_iter {
	u8 iter_type;
	bool copy_mc;
	bool nofault;
	bool data_source;
	size_t iov_offset;
	/*
	 * Hack alert: overlay ubuf_iovec with iovec + count, so
	 * that the members resolve correctly regardless of the type
	 * of iterator used. This means that you can use:
	 *
	 * &iter->__ubuf_iovec or iter->__iov
	 *
	 * interchangably for the user_backed cases, hence simplifying
	 * some of the cases that need to deal with both.
	 */
	union {
		/*
		 * This really should be a const, but we cannot do that without
		 * also modifying any of the zero-filling iter init functions.
		 * Leave it non-const for now, but it should be treated as such.
		 */
		struct iovec __ubuf_iovec;
		struct {
			union {
				/* use iter_iov() to get the current vec */
				const struct iovec *__iov;
				const struct kvec *kvec;
				const struct bio_vec *bvec;
				struct xarray *xarray;
				void __user *ubuf;
			};
			size_t count;
		};
	};
	union {
		unsigned long nr_segs;
		loff_t xarray_start;
	};
};

• iter_type : 내부적으로 어떤 구조체를 사용하는지에 대한 정보이다. enum iter_type의 값들이 저장된다.
• copy_mc :
• nofault :
• data_source : read / write 중 어떤 연산을 위한 것인지에 대한 정보이다. ITER_SOURCE ( == WRITE), ITER_DEST ( == READ)의 값이 저장된다.
• iov_offset : 현재 어디까지 read/write가 되었는지에 대한 offset이다.
• first union : 실제 데이터가 저장된 vector들이 담겨있다. 또한, count변수에 버퍼들의 총 size의 합이 저장된다.
• second union : vector의 크기를 나타낸다.

sys_readv()의 소스 코드를 통해 자세히 알아보자.

sys_readv()

SYSCALL_DEFINE3(readv, unsigned long, fd, const struct iovec __user *, vec,
		unsigned long, vlen)
{
	return do_readv(fd, vec, vlen, 0);
}

static ssize_t do_readv(unsigned long fd, const struct iovec __user *vec,
			unsigned long vlen, rwf_t flags)
{
	struct fd f = fdget_pos(fd);
	ssize_t ret = -EBADF;

	if (f.file) {
		loff_t pos, *ppos = file_ppos(f.file);
		if (ppos) {
			pos = *ppos;
			ppos = &pos;
		}
		ret = vfs_readv(f.file, vec, vlen, ppos, flags);
		if (ret >= 0 && ppos)
			f.file->f_pos = pos;
		fdput_pos(f);
	}

	if (ret > 0)
		add_rchar(current, ret);
	inc_syscr(current);
	return ret;
}

크게 중요한 내용은 없다.
sys_readv() -> do_readv() -> vfs_readv() 순으로 호출된다.

static ssize_t vfs_readv(struct file *file, const struct iovec __user *vec,
		  unsigned long vlen, loff_t *pos, rwf_t flags)
{
	struct iovec iovstack[UIO_FASTIOV];
	struct iovec *iov = iovstack;
	struct iov_iter iter;
	ssize_t ret;

	ret = import_iovec(ITER_DEST, vec, vlen, ARRAY_SIZE(iovstack), &iov, &iter);
	if (ret >= 0) {
		ret = do_iter_read(file, &iter, pos, flags);
		kfree(iov);
	}

	return ret;
}

iovstack은 빠른 접근을 위한 스택의 iovec 배열에 해당한다.
우선적으로 iov 변수를 iovstack을 가리키도록 한다.

import_iovec()을 통해 iovec, iov_iter를 초기화한다. 이 때, 인자로 ITER_DEST를 전달함으로써 read관련 연산을 할 것이라는 정보를 같이 준다.

do_iter_read()를 통해 실제 read를 수행하게 된다.

import_iovec()

/**
 * import_iovec() - Copy an array of &struct iovec from userspace
 *     into the kernel, check that it is valid, and initialize a new
 *     &struct iov_iter iterator to access it.
 *
 * @type: One of %READ or %WRITE.
 * @uvec: Pointer to the userspace array.
 * @nr_segs: Number of elements in userspace array.
 * @fast_segs: Number of elements in @iov.
 * @iovp: (input and output parameter) Pointer to pointer to (usually small
 *     on-stack) kernel array.
 * @i: Pointer to iterator that will be initialized on success.
 *
 * If the array pointed to by *@iov is large enough to hold all @nr_segs,
 * then this function places %NULL in *@iov on return. Otherwise, a new
 * array will be allocated and the result placed in *@iov. This means that
 * the caller may call kfree() on *@iov regardless of whether the small
 * on-stack array was used or not (and regardless of whether this function
 * returns an error or not).
 *
 * Return: Negative error code on error, bytes imported on success
 */
ssize_t import_iovec(int type, const struct iovec __user *uvec,
		 unsigned nr_segs, unsigned fast_segs,
		 struct iovec **iovp, struct iov_iter *i)
{
	return __import_iovec(type, uvec, nr_segs, fast_segs, iovp, i,
			      in_compat_syscall());
}
EXPORT_SYMBOL(import_iovec);

주석에 설명이 잘 나와있다.
userspace에서 전달한 iovec을 kernel로 복사해 오고, valid한지 체크한다.
그 후 iov_iter를 초기화하는 함수이다.

앞서 vfs_readv()에서의 iovstack 값을 우선적으로 사용하는데, 만약 nr_segs의 값이 iovstack의 크기 (fast_segs)보다 클 경우 새로 iovec을 할당받아서 사용하게 된다.

ssize_t __import_iovec(int type, const struct iovec __user *uvec,
		 unsigned nr_segs, unsigned fast_segs, struct iovec **iovp,
		 struct iov_iter *i, bool compat)
{
	ssize_t total_len = 0;
	unsigned long seg;
	struct iovec *iov;

	if (nr_segs == 1)
		return __import_iovec_ubuf(type, uvec, iovp, i, compat);

	iov = iovec_from_user(uvec, nr_segs, fast_segs, *iovp, compat);
	if (IS_ERR(iov)) {
		*iovp = NULL;
		return PTR_ERR(iov);
	}

	/*
	 * According to the Single Unix Specification we should return EINVAL if
	 * an element length is < 0 when cast to ssize_t or if the total length
	 * would overflow the ssize_t return value of the system call.
	 *
	 * Linux caps all read/write calls to MAX_RW_COUNT, and avoids the
	 * overflow case.
	 */
	for (seg = 0; seg < nr_segs; seg++) {
		ssize_t len = (ssize_t)iov[seg].iov_len;

		if (!access_ok(iov[seg].iov_base, len)) {
			if (iov != *iovp)
				kfree(iov);
			*iovp = NULL;
			return -EFAULT;
		}

		if (len > MAX_RW_COUNT - total_len) {
			len = MAX_RW_COUNT - total_len;
			iov[seg].iov_len = len;
		}
		total_len += len;
	}

	iov_iter_init(i, type, iov, nr_segs, total_len);
	if (iov == *iovp)
		*iovp = NULL;
	else
		*iovp = iov;
	return total_len;
}

만약 nr_segs의 값이 1인 경우, __import_iovec_ubuf()를 호출한 후 바로 리턴한다.

그 후 iovec_from_user()를 통해 iovec을 user로 부터 복사해 온 후, 각 iovec에 대해 access가 가능한지 확인한다.

마지막으로 iov_iter_init()을 호출하여 iovec에 맞게끔 초기화한다.

struct iovec *iovec_from_user(const struct iovec __user *uvec,
		unsigned long nr_segs, unsigned long fast_segs,
		struct iovec *fast_iov, bool compat)
{
	struct iovec *iov = fast_iov;
	int ret;

	/*
	 * SuS says "The readv() function *may* fail if the iovcnt argument was
	 * less than or equal to 0, or greater than {IOV_MAX}.  Linux has
	 * traditionally returned zero for zero segments, so...
	 */
	if (nr_segs == 0)
		return iov;
	if (nr_segs > UIO_MAXIOV)
		return ERR_PTR(-EINVAL);
	if (nr_segs > fast_segs) {
		iov = kmalloc_array(nr_segs, sizeof(struct iovec), GFP_KERNEL);
		if (!iov)
			return ERR_PTR(-ENOMEM);
	}

	if (unlikely(compat))
		ret = copy_compat_iovec_from_user(iov, uvec, nr_segs);
	else
		ret = copy_iovec_from_user(iov, uvec, nr_segs);
	if (ret) {
		if (iov != fast_iov)
			kfree(iov);
		return ERR_PTR(ret);
	}

	return iov;
}

여기서 iovstack을 사용할지의 여부가 결정된다.
만약 nr_segs > fast_segs인 경우, 새롭게 iov에 값을 할당해 준다.
그렇지 않다면 그대로 iovstack을 사용한다.

그 다음에 copy_iovec_from_user()를 통해 userland로 부터 iovec을 복사해 온다.

static __noclone int copy_iovec_from_user(struct iovec *iov,
		const struct iovec __user *uiov, unsigned long nr_segs)
{
	int ret = -EFAULT;

	if (!user_access_begin(uiov, nr_segs * sizeof(*uiov)))
		return -EFAULT;

	do {
		void __user *buf;
		ssize_t len;

		unsafe_get_user(len, &uiov->iov_len, uaccess_end);
		unsafe_get_user(buf, &uiov->iov_base, uaccess_end);

		/* check for size_t not fitting in ssize_t .. */
		if (unlikely(len < 0)) {
			ret = -EINVAL;
			goto uaccess_end;
		}
		iov->iov_base = buf;
		iov->iov_len = len;

		uiov++; iov++;
	} while (--nr_segs);

	ret = 0;
uaccess_end:
	user_access_end();
	return ret;
}

nr_segs 만큼 while문을 돌면서 userland에서 kernel로 값을 복사해 오는 함수이다.

void iov_iter_init(struct iov_iter *i, unsigned int direction,
			const struct iovec *iov, unsigned long nr_segs,
			size_t count)
{
	WARN_ON(direction & ~(READ | WRITE));
	*i = (struct iov_iter) {
		.iter_type = ITER_IOVEC,
		.copy_mc = false,
		.nofault = false,
		.data_source = direction,
		.__iov = iov,
		.nr_segs = nr_segs,
		.iov_offset = 0,
		.count = count
	};
}
EXPORT_SYMBOL(iov_iter_init);

얘도 별로 중요한 건 없다.
단순히 iov에 저장된 값을 기반으로 iov_iter를 초기화하는 함수이다.

지금까지의 과정을 통해 iov_iter의 값이 초기화가 되었다.
이제 본격적으로 read를 시작하는 do_iter_read()에 대해 살펴보자.

do_iter_read()

static ssize_t do_iter_read(struct file *file, struct iov_iter *iter,
		loff_t *pos, rwf_t flags)
{
	size_t tot_len;
	ssize_t ret = 0;

	if (!(file->f_mode & FMODE_READ))
		return -EBADF;
	if (!(file->f_mode & FMODE_CAN_READ))
		return -EINVAL;

	tot_len = iov_iter_count(iter);
	if (!tot_len)
		goto out;
	ret = rw_verify_area(READ, file, pos, tot_len);
	if (ret < 0)
		return ret;

	if (file->f_op->read_iter)
		ret = do_iter_readv_writev(file, iter, pos, READ, flags);
	else
		ret = do_loop_readv_writev(file, iter, pos, READ, flags);
out:
	if (ret >= 0)
		fsnotify_access(file);
	return ret;
}

여러 접근 권한에 대해 먼저 체크를 한다.
그 후 file->f_op->read_iter에 함수가 저장되어 있는지 확인하는데, 만약 저장되어 있다면 do_iter_readv_writev()를, 그렇지 않다면 do_loop_readv_writev()를 호출하게 된다.

static ssize_t do_iter_readv_writev(struct file *filp, struct iov_iter *iter,
		loff_t *ppos, int type, rwf_t flags)
{
	struct kiocb kiocb;
	ssize_t ret;

	init_sync_kiocb(&kiocb, filp);
	ret = kiocb_set_rw_flags(&kiocb, flags);
	if (ret)
		return ret;
	kiocb.ki_pos = (ppos ? *ppos : 0);

	if (type == READ)
		ret = call_read_iter(filp, &kiocb, iter);
	else
		ret = call_write_iter(filp, &kiocb, iter);
	BUG_ON(ret == -EIOCBQUEUED);
	if (ppos)
		*ppos = kiocb.ki_pos;
	return ret;
}

kiocb 구조체를 초기화 한 후, call_read_iter()를 호출하게 된다.
kiocb에 대해서는 나중에 정리해 보고자 한다.

static inline ssize_t call_read_iter(struct file *file, struct kiocb *kio,
				     struct iov_iter *iter)
{
	return file->f_op->read_iter(kio, iter);
}

file->f_op에 저장된 read_iter() 함수를 호출한다.

/* Do it by hand, with file-ops */
static ssize_t do_loop_readv_writev(struct file *filp, struct iov_iter *iter,
		loff_t *ppos, int type, rwf_t flags)
{
	ssize_t ret = 0;

	if (flags & ~RWF_HIPRI)
		return -EOPNOTSUPP;

	while (iov_iter_count(iter)) {
		ssize_t nr;

		if (type == READ) {
			nr = filp->f_op->read(filp, iter_iov_addr(iter),
						iter_iov_len(iter), ppos);
		} else {
			nr = filp->f_op->write(filp, iter_iov_addr(iter),
						iter_iov_len(iter), ppos);
		}

		if (nr < 0) {
			if (!ret)
				ret = nr;
			break;
		}
		ret += nr;
		if (nr != iter_iov_len(iter))
			break;
		iov_iter_advance(iter, nr);
	}

	return ret;
}

이 함수가 호출된 경우는 f_op->read_iter가 NULL인 경우이므로, 실제로 while문을 돌면서 각각의 버퍼에 대해 f_op->read를 호출해 준다.

이 때 쓰이는 iov_iter_*() 함수들에 대해 간단히만 알아보도록 하자.

iov_iter_count()

static inline size_t iov_iter_count(const struct iov_iter *i)
{
	return i->count;
}

단순히 남은 길이를 리턴한다.

iter_iov_addr/len()

static inline const struct iovec *iter_iov(const struct iov_iter *iter)
{
	if (iter->iter_type == ITER_UBUF)
		return (const struct iovec *) &iter->__ubuf_iovec;
	return iter->__iov;
}

#define iter_iov_addr(iter)	(iter_iov(iter)->iov_base + (iter)->iov_offset)
#define iter_iov_len(iter)	(iter_iov(iter)->iov_len - (iter)->iov_offset)

• iter_iov_addr()
  다음 read/write를 시작하는 위치에 대해 리턴하는 매크로로, 풀어쓰면 iter->__iov->iov_base + iter->iov_offset이 된다.
• iter_iov_len()
  말 그대로 read/write 할 길이를 리턴하는 매크로로, 풀어쓰면 iter->__iov->iov_len - iter->iov_offset이 된다.

iov_iter_advance

void iov_iter_advance(struct iov_iter *i, size_t size)
{
	if (unlikely(i->count < size))
		size = i->count;
	if (likely(iter_is_ubuf(i)) || unlikely(iov_iter_is_xarray(i))) {
		i->iov_offset += size;
		i->count -= size;
	} else if (likely(iter_is_iovec(i) || iov_iter_is_kvec(i))) {
		/* iovec and kvec have identical layouts */
		iov_iter_iovec_advance(i, size);
	} else if (iov_iter_is_bvec(i)) {
		iov_iter_bvec_advance(i, size);
	} else if (iov_iter_is_discard(i)) {
		i->count -= size;
	}
}

특정한 길이만큼 read/write가 끝난 후 다음 read/write를 위해 iov_iter 내부의 상태를 업데이트를 해 주는 함수이다.

iov_iter가 여러 구조체들을 다루는 만큼 각각의 경우에 따라 다른 함수가 호출된다.
우리는 iovec에 대해서만 보고 있으므로, iov_iter_iovec_advance()에 대해서만 살펴본다.

static void iov_iter_iovec_advance(struct iov_iter *i, size_t size)
{
	const struct iovec *iov, *end;

	if (!i->count)
		return;
	i->count -= size;

	size += i->iov_offset; // from beginning of current segment
	for (iov = iter_iov(i), end = iov + i->nr_segs; iov < end; iov++) {
		if (likely(size < iov->iov_len))
			break;
		size -= iov->iov_len;
	}
	i->iov_offset = size;
	i->nr_segs -= iov - iter_iov(i);
	i->__iov = iov;
}

iov_iter의 동작 원리가 지금까지 read/write해 온 결과는 버리고 다음 결과부터 시작 지점에 저장되도록 동작하는 것을 확인할 수 있다.

사실 writev() 의 경우, 코드를 보면 알겠지만 readv_writev() 형태로 묶여있는 경우가 많았던 것처럼 동작이 유사하므로 생략한다.

Reference

• https://elixir.bootlin.com/linux/v6.7.9/source
• https://man7.org/linux/man-pages/man2/writev.2.html