cmds/incidentd/src/FdBuffer.cpp

/*
 * Copyright (C) 2016 The Android Open Source Project
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#include "Log.h"

#include "FdBuffer.h"

#include <cutils/log.h>
#include <utils/SystemClock.h>

#include <fcntl.h>
#include <poll.h>
#include <unistd.h>
#include <wait.h>

const ssize_t BUFFER_SIZE = 16 * 1024;  // 16 KB
const ssize_t MAX_BUFFER_COUNT = 256;   // 4 MB max

FdBuffer::FdBuffer()
    : mBuffer(BUFFER_SIZE), mStartTime(-1), mFinishTime(-1), mTimedOut(false), mTruncated(false) {}

FdBuffer::~FdBuffer() {}

status_t FdBuffer::read(int fd, int64_t timeout) {
    struct pollfd pfds = {.fd = fd, .events = POLLIN};
    mStartTime = uptimeMillis();

    fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);

    while (true) {
        if (mBuffer.size() >= MAX_BUFFER_COUNT * BUFFER_SIZE) {
            mTruncated = true;
            break;
        }
        if (mBuffer.writeBuffer() == NULL) return NO_MEMORY;

        int64_t remainingTime = (mStartTime + timeout) - uptimeMillis();
        if (remainingTime <= 0) {
            VLOG("timed out due to long read");
            mTimedOut = true;
            break;
        }

        int count = poll(&pfds, 1, remainingTime);
        if (count == 0) {
            VLOG("timed out due to block calling poll");
            mTimedOut = true;
            break;
        } else if (count < 0) {
            VLOG("poll failed: %s", strerror(errno));
            return -errno;
        } else {
            if ((pfds.revents & POLLERR) != 0) {
                VLOG("return event has error %s", strerror(errno));
                return errno != 0 ? -errno : UNKNOWN_ERROR;
            } else {
                ssize_t amt = ::read(fd, mBuffer.writeBuffer(), mBuffer.currentToWrite());
                if (amt < 0) {
                    if (errno == EAGAIN || errno == EWOULDBLOCK) {
                        continue;
                    } else {
                        VLOG("Fail to read %d: %s", fd, strerror(errno));
                        return -errno;
                    }
                } else if (amt == 0) {
                    break;
                }
                mBuffer.wp()->move(amt);
            }
        }
    }
    mFinishTime = uptimeMillis();
    return NO_ERROR;
}

status_t FdBuffer::readProcessedDataInStream(int fd, int toFd, int fromFd, int64_t timeoutMs,
                                             const bool isSysfs) {
    struct pollfd pfds[] = {
            {.fd = fd, .events = POLLIN},
            {.fd = toFd, .events = POLLOUT},
            {.fd = fromFd, .events = POLLIN},
    };

    mStartTime = uptimeMillis();

    // mark all fds non blocking
    fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
    fcntl(toFd, F_SETFL, fcntl(toFd, F_GETFL, 0) | O_NONBLOCK);
    fcntl(fromFd, F_SETFL, fcntl(fromFd, F_GETFL, 0) | O_NONBLOCK);

    // A circular buffer holds data read from fd and writes to parsing process
    uint8_t cirBuf[BUFFER_SIZE];
    size_t cirSize = 0;
    int rpos = 0, wpos = 0;

    // This is the buffer used to store processed data
    while (true) {
        if (mBuffer.size() >= MAX_BUFFER_COUNT * BUFFER_SIZE) {
            mTruncated = true;
            break;
        }
        if (mBuffer.writeBuffer() == NULL) return NO_MEMORY;

        int64_t remainingTime = (mStartTime + timeoutMs) - uptimeMillis();
        if (remainingTime <= 0) {
            VLOG("timed out due to long read");
            mTimedOut = true;
            break;
        }

        // wait for any pfds to be ready to perform IO
        int count = poll(pfds, 3, remainingTime);
        if (count == 0) {
            VLOG("timed out due to block calling poll");
            mTimedOut = true;
            break;
        } else if (count < 0) {
            VLOG("Fail to poll: %s", strerror(errno));
            return -errno;
        }

        // make sure no errors occur on any fds
        for (int i = 0; i < 3; ++i) {
            if ((pfds[i].revents & POLLERR) != 0) {
                if (i == 0 && isSysfs) {
                    VLOG("fd %d is sysfs, ignore its POLLERR return value", fd);
                    continue;
                }
                VLOG("fd[%d]=%d returns error events: %s", i, fd, strerror(errno));
                return errno != 0 ? -errno : UNKNOWN_ERROR;
            }
        }

        // read from fd
        if (cirSize != BUFFER_SIZE && pfds[0].fd != -1) {
            ssize_t amt;
            if (rpos >= wpos) {
                amt = ::read(fd, cirBuf + rpos, BUFFER_SIZE - rpos);
            } else {
                amt = ::read(fd, cirBuf + rpos, wpos - rpos);
            }
            if (amt < 0) {
                if (!(errno == EAGAIN || errno == EWOULDBLOCK)) {
                    VLOG("Fail to read fd %d: %s", fd, strerror(errno));
                    return -errno;
                }                   // otherwise just continue
            } else if (amt == 0) {  // reach EOF so don't have to poll pfds[0].
                ::close(pfds[0].fd);
                pfds[0].fd = -1;
            } else {
                rpos += amt;
                cirSize += amt;
            }
        }

        // write to parsing process
        if (cirSize > 0 && pfds[1].fd != -1) {
            ssize_t amt;
            if (rpos > wpos) {
                amt = ::write(toFd, cirBuf + wpos, rpos - wpos);
            } else {
                amt = ::write(toFd, cirBuf + wpos, BUFFER_SIZE - wpos);
            }
            if (amt < 0) {
                if (!(errno == EAGAIN || errno == EWOULDBLOCK)) {
                    VLOG("Fail to write toFd %d: %s", toFd, strerror(errno));
                    return -errno;
                }  // otherwise just continue
            } else {
                wpos += amt;
                cirSize -= amt;
            }
        }

        // if buffer is empty and fd is closed, close write fd.
        if (cirSize == 0 && pfds[0].fd == -1 && pfds[1].fd != -1) {
            ::close(pfds[1].fd);
            pfds[1].fd = -1;
        }

        // circular buffer, reset rpos and wpos
        if (rpos >= BUFFER_SIZE) {
            rpos = 0;
        }
        if (wpos >= BUFFER_SIZE) {
            wpos = 0;
        }

        // read from parsing process
        ssize_t amt = ::read(fromFd, mBuffer.writeBuffer(), mBuffer.currentToWrite());
        if (amt < 0) {
            if (!(errno == EAGAIN || errno == EWOULDBLOCK)) {
                VLOG("Fail to read fromFd %d: %s", fromFd, strerror(errno));
                return -errno;
            }  // otherwise just continue
        } else if (amt == 0) {
            break;
        } else {
            mBuffer.wp()->move(amt);
        }
    }

    mFinishTime = uptimeMillis();
    return NO_ERROR;
}

size_t FdBuffer::size() const { return mBuffer.size(); }

EncodedBuffer::iterator FdBuffer::data() const { return mBuffer.begin(); }