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.
 */
#define DEBUG false
#include "Log.h"

#include "FdBuffer.h"

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

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

namespace android {
namespace os {
namespace incidentd {

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

FdBuffer::FdBuffer()
        :mBuffer(new EncodedBuffer(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;
            VLOG("Truncating data");
            break;
        }
        if (mBuffer->writeBuffer() == NULL) {
            VLOG("No memory");
            return NO_MEMORY;
        }

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

        int count = TEMP_FAILURE_RETRY(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 = TEMP_FAILURE_RETRY(
                        ::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) {
                    VLOG("Reached EOF of fd=%d", fd);
                    break;
                }
                mBuffer->wp()->move(amt);
            }
        }
    }
    mFinishTime = uptimeMillis();
    return NO_ERROR;
}

status_t FdBuffer::readFully(int fd) {
    mStartTime = uptimeMillis();

    while (true) {
        if (mBuffer->size() >= MAX_BUFFER_COUNT * BUFFER_SIZE) {
            // Don't let it get too big.
            mTruncated = true;
            VLOG("Truncating data");
            break;
        }
        if (mBuffer->writeBuffer() == NULL) {
            VLOG("No memory");
            return NO_MEMORY;
        }

        ssize_t amt =
                TEMP_FAILURE_RETRY(::read(fd, mBuffer->writeBuffer(), mBuffer->currentToWrite()));
        if (amt < 0) {
            VLOG("Fail to read %d: %s", fd, strerror(errno));
            return -errno;
        } else if (amt == 0) {
            VLOG("Done reading %zu bytes", mBuffer->size());
            // We're done.
            break;
        }
        mBuffer->wp()->move(amt);
    }

    mFinishTime = uptimeMillis();
    return NO_ERROR;
}

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

    mStartTime = uptimeMillis();

    // mark all fds non blocking
    fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
    fcntl(toFd.get(), F_SETFL, fcntl(toFd.get(), F_GETFL, 0) | O_NONBLOCK);
    fcntl(fromFd.get(), F_SETFL, fcntl(fromFd.get(), 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) {
            VLOG("Truncating data");
            mTruncated = true;
            break;
        }
        if (mBuffer->writeBuffer() == NULL) {
            VLOG("No memory");
            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 = TEMP_FAILURE_RETRY(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 = TEMP_FAILURE_RETRY(::read(fd, cirBuf + rpos, BUFFER_SIZE - rpos));
            } else {
                amt = TEMP_FAILURE_RETRY(::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) {
                VLOG("Reached EOF of input file %d", fd);
                pfds[0].fd = -1;  // reach EOF so don't have to poll pfds[0].
            } else {
                rpos += amt;
                cirSize += amt;
            }
        }

        // write to parsing process
        if (cirSize > 0 && pfds[1].fd != -1) {
            ssize_t amt;
            if (rpos > wpos) {
                amt = TEMP_FAILURE_RETRY(::write(toFd.get(), cirBuf + wpos, rpos - wpos));
            } else {
                amt = TEMP_FAILURE_RETRY(::write(toFd.get(), cirBuf + wpos, BUFFER_SIZE - wpos));
            }
            if (amt < 0) {
                if (!(errno == EAGAIN || errno == EWOULDBLOCK)) {
                    VLOG("Fail to write toFd %d: %s", toFd.get(), 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) {
            VLOG("Close write pipe %d", toFd.get());
            toFd.reset();
            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 = TEMP_FAILURE_RETRY(
                ::read(fromFd.get(), mBuffer->writeBuffer(), mBuffer->currentToWrite()));
        if (amt < 0) {
            if (!(errno == EAGAIN || errno == EWOULDBLOCK)) {
                VLOG("Fail to read fromFd %d: %s", fromFd.get(), strerror(errno));
                return -errno;
            }  // otherwise just continue
        } else if (amt == 0) {
            VLOG("Reached EOF of fromFd %d", fromFd.get());
            break;
        } else {
            mBuffer->wp()->move(amt);
        }
    }

    mFinishTime = uptimeMillis();
    return NO_ERROR;
}

status_t FdBuffer::write(uint8_t const* buf, size_t size) {
    return mBuffer->writeRaw(buf, size);
}

status_t FdBuffer::write(const sp<ProtoReader>& reader) {
    return mBuffer->writeRaw(reader);
}

status_t FdBuffer::write(const sp<ProtoReader>& reader, size_t size) {
    return mBuffer->writeRaw(reader, size);
}

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

sp<EncodedBuffer> FdBuffer::data() const {
    return mBuffer;
}

}  // namespace incidentd
}  // namespace os
}  // namespace android