shttpd/task/sockrw.c

/* shttpd - IO on stream sockets
   Copyright (C) 2018 Ariadne Devos

   This program is free software: you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation, either version 3 of the License, or
   (at your option) any later version.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program.  If not, see <http://www.gnu.org/licenses/>. */

#include "fd.h"
#include <sHT/bugs.h>
#include <sHT/compiler.h>
#include <sHT/nospec.h>
#include <sHT/stream.h>
#include "worker.h"

#include <errno.h>
#include <stddef.h>
#include <stdint.h>
#include <sys/epoll.h>
#include <sys/socket.h>

/* These are sorted in order of expected prevalence */
enum sHT_send_err_type {
        sHT_SEND_BLOCKING,
        sHT_SEND_GRACEFUL_RESET,
        sHT_SEND_BLUNT_RESET,
        sHT_SEND_GRACEFUL_CLOSE,
        /* like in EINTR */
        sHT_SEND_INTERRUPTED,
        /* e.g. timeout, connection reset */
        sHT_SEND_TRANSIENT,
        sHT_SEND_KERNEL_OOM,
        /* Anything we didn't expect.
           (We expect malicious clients,
           but no sHT bugs.)

           This should be logged as a warning. */
        sHT_SEND_OTHER,
};

__attribute__((const))
static enum sHT_send_err_type
sHT_classify_sentrecv_tcp(int err)
{
	switch (err) {
#if EWOULDBLOCK != EAGAIN
	case EWOULDBLOCK: /* fallthrough */
#endif
	case EAGAIN:
		return sHT_SEND_BLOCKING;
	case EINTR:
		return sHT_SEND_INTERRUPTED;
	case ECONNRESET:
		return sHT_SEND_GRACEFUL_RESET;
	case EPIPE:
		return sHT_SEND_GRACEFUL_CLOSE;
	case ETIMEDOUT: /* fallthrough */
	case EHOSTUNREACH:
		return sHT_SEND_BLUNT_RESET;
	case ENOBUFS:
		/* no busy loops? */
		return sHT_SEND_TRANSIENT;
	case ENOMEM:
		return sHT_SEND_KERNEL_OOM;
	default:
		return sHT_SEND_OTHER;
	}
}

/* True if it should be retried directly, false otherwise. */
__attribute__((nonnull (1, 2)))
static _Bool
sHT_socket_sendrecv_errno(struct sHT_worker *worker, struct sHT_task_stream *task, int err, uint32_t epollflags)
{
	/* XXX use err, not errno */
	switch (sHT_classify_sentrecv_tcp(errno)) {
	case sHT_SEND_BLOCKING:
		task->task.epollflags &= ~epollflags;
		return 0;
	case sHT_SEND_GRACEFUL_CLOSE:
		task->stream.flags |= sHT_STREAM_WRITE_EOF;
		task->task.epollflags &= ~epollflags;
		return 0;
	case sHT_SEND_GRACEFUL_RESET:
		task->stream.flags |= sHT_STREAM_WRITE_EOF | sHT_STREAM_READ_EOF | sHT_STREAM_RESET_GRACEFUL;
		task->task.epollflags &= ~(EPOLLIN | EPOLLOUT);
		return 0;
	case sHT_SEND_BLUNT_RESET:
		task->stream.flags |= sHT_STREAM_WRITE_EOF | sHT_STREAM_READ_EOF | sHT_STREAM_RESET_BLUNT;
		task->task.epollflags &= ~(EPOLLIN | EPOLLOUT);
		return 0;
	case sHT_SEND_INTERRUPTED:
		return 1;
	case sHT_SEND_TRANSIENT:
		/* TODO: may be a good idea to log these too,
		   as an informational message */
		sHT_queue_task(worker->todo, &task->task);
		return 0;
	case sHT_SEND_KERNEL_OOM:
		/* No, I don't like overcommiting.
		   Killing is better than hanging, though. */
		worker->flags |= sHT_WORKER_OOM;
		sHT_queue_task(worker->todo, &task->task);
		return 0;
	case sHT_SEND_OTHER:
		sHT_todo("didn't recognise TCP error");
	default:
		sHT_assert(0);
	}
}

void
sHT_socket_sendsome_tcp(struct sHT_worker *worker, struct sHT_task_stream *task)
{
	const unsigned char *buf = task->stream.to_write.first;
	size_t start = task->stream.to_write.offset;
	size_t end = (task->stream.to_write.offset + task->stream.to_write.length) % sHT_PAPER_SIZE;
	if (end < start)
		end = sHT_PAPER_SIZE;
	end = sHT_index_nospec(end, sHT_PAPER_SIZE - start);
	do {
		/* XXX: speculatively negative sizes? */
		ssize_t sent = send(task->stream.fd, buf + start, end - start, MSG_DONTWAIT | MSG_NOSIGNAL);
		if (sent < 0)
			continue;
		/* some data is on the kernel queue, or the NIC ... */
		sHT_assert(sent <= task->stream.to_write.length);
		task->stream.to_write.offset = (task->stream.to_write.offset + sent) % sHT_PAPER_SIZE;
		task->stream.to_write.length -= sent;
		return;
	} while (sHT_unlikely(sHT_socket_sendrecv_errno(worker, task, errno, EPOLLOUT)));
}

void
sHT_socket_readsome_tcp(struct sHT_worker *worker, struct sHT_task_stream *task)
{
	unsigned char *buf = task->stream.has_read.first;
	size_t start = (task->stream.has_read.offset + task->stream.has_read.length) % sHT_PAPER_SIZE;
	size_t end = task->stream.has_read.offset;
	if (end < start)
		end = sHT_PAPER_SIZE;
	end = sHT_index_nospec(end, sHT_PAPER_SIZE - start);
	/* XXX: speculatively negative sizes? */
	do {
		ssize_t received;
		received = recv(task->stream.fd, buf + start, end - start, MSG_DONTWAIT);
		if (received < 0)
			continue;
		sHT_assert(received <= sHT_PAPER_SIZE - task->stream.has_read.length);
		task->stream.has_read.length += received;
		return;
	} while (sHT_unlikely(sHT_socket_sendrecv_errno(worker, task, errno, EPOLLOUT)));
}