static int s_process_write_requests()

in source/posix/socket.c [1448:1577]

• 108 lines of code
• 13 McCabe index (conditional complexity)

static int s_process_write_requests(struct aws_socket *socket, struct write_request *parent_request) {
    struct posix_socket *socket_impl = socket->impl;

    if (parent_request) {
        AWS_LOGF_TRACE(
            AWS_LS_IO_SOCKET,
            "id=%p fd=%d: processing write requests, called from aws_socket_write",
            (void *)socket,
            socket->io_handle.data.fd);
    } else {
        AWS_LOGF_TRACE(
            AWS_LS_IO_SOCKET,
            "id=%p fd=%d: processing write requests, invoked by the event-loop",
            (void *)socket,
            socket->io_handle.data.fd);
    }

    bool purge = false;
    int aws_error = AWS_OP_SUCCESS;
    bool parent_request_failed = false;
    bool pushed_to_written_queue = false;

    /* if a close call happens in the middle, this queue will have been cleaned out from under us. */
    while (!aws_linked_list_empty(&socket_impl->write_queue)) {
        struct aws_linked_list_node *node = aws_linked_list_front(&socket_impl->write_queue);
        struct write_request *write_request = AWS_CONTAINER_OF(node, struct write_request, node);

        AWS_LOGF_TRACE(
            AWS_LS_IO_SOCKET,
            "id=%p fd=%d: dequeued write request of size %llu, remaining to write %llu",
            (void *)socket,
            socket->io_handle.data.fd,
            (unsigned long long)write_request->original_buffer_len,
            (unsigned long long)write_request->cursor_cpy.len);

        ssize_t written = send(
            socket->io_handle.data.fd, write_request->cursor_cpy.ptr, write_request->cursor_cpy.len, NO_SIGNAL_SEND);

        AWS_LOGF_TRACE(
            AWS_LS_IO_SOCKET,
            "id=%p fd=%d: send written size %d",
            (void *)socket,
            socket->io_handle.data.fd,
            (int)written);

        if (written < 0) {
            int error = errno;
            if (error == EAGAIN) {
                AWS_LOGF_TRACE(
                    AWS_LS_IO_SOCKET, "id=%p fd=%d: returned would block", (void *)socket, socket->io_handle.data.fd);
                break;
            }

            if (error == EPIPE) {
                AWS_LOGF_DEBUG(
                    AWS_LS_IO_SOCKET,
                    "id=%p fd=%d: already closed before write",
                    (void *)socket,
                    socket->io_handle.data.fd);
                aws_error = AWS_IO_SOCKET_CLOSED;
                aws_raise_error(aws_error);
                purge = true;
                break;
            }

            purge = true;
            AWS_LOGF_DEBUG(
                AWS_LS_IO_SOCKET,
                "id=%p fd=%d: write error with error code %d",
                (void *)socket,
                socket->io_handle.data.fd,
                error);
            aws_error = s_determine_socket_error(error);
            aws_raise_error(aws_error);
            break;
        }

        size_t remaining_to_write = write_request->cursor_cpy.len;

        aws_byte_cursor_advance(&write_request->cursor_cpy, (size_t)written);
        AWS_LOGF_TRACE(
            AWS_LS_IO_SOCKET,
            "id=%p fd=%d: remaining write request to write %llu",
            (void *)socket,
            socket->io_handle.data.fd,
            (unsigned long long)write_request->cursor_cpy.len);

        if ((size_t)written == remaining_to_write) {
            AWS_LOGF_TRACE(
                AWS_LS_IO_SOCKET, "id=%p fd=%d: write request completed", (void *)socket, socket->io_handle.data.fd);

            aws_linked_list_remove(node);
            write_request->error_code = AWS_ERROR_SUCCESS;
            aws_linked_list_push_back(&socket_impl->written_queue, node);
            pushed_to_written_queue = true;
        }
    }

    if (purge) {
        while (!aws_linked_list_empty(&socket_impl->write_queue)) {
            struct aws_linked_list_node *node = aws_linked_list_pop_front(&socket_impl->write_queue);
            struct write_request *write_request = AWS_CONTAINER_OF(node, struct write_request, node);

            /* If this fn was invoked directly from aws_socket_write(), don't invoke the error callback
             * as the user will be able to rely on the return value from aws_socket_write() */
            if (write_request == parent_request) {
                parent_request_failed = true;
                aws_mem_release(socket->allocator, write_request);
            } else {
                write_request->error_code = aws_error;
                aws_linked_list_push_back(&socket_impl->written_queue, node);
                pushed_to_written_queue = true;
            }
        }
    }

    if (pushed_to_written_queue && !socket_impl->written_task_scheduled) {
        socket_impl->written_task_scheduled = true;
        aws_task_init(&socket_impl->written_task, s_written_task, socket, "socket_written_task");
        aws_event_loop_schedule_task_now(socket->event_loop, &socket_impl->written_task);
    }

    /* Only report error if aws_socket_write() invoked this function and its write_request failed */
    if (!parent_request_failed) {
        return AWS_OP_SUCCESS;
    }

    aws_raise_error(aws_error);
    return AWS_OP_ERR;
}