/*
* 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 = 6144; // 96 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