1/*
2 * Copyright (C) 2014 The Android Open Source Project
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 *      http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17#include <errno.h>
18#include <sys/prctl.h>
19
20#include "FlushCommand.h"
21#include "LogBuffer.h"
22#include "LogTimes.h"
23#include "LogReader.h"
24
25pthread_mutex_t LogTimeEntry::timesLock = PTHREAD_MUTEX_INITIALIZER;
26
27LogTimeEntry::LogTimeEntry(LogReader &reader, SocketClient *client,
28                           bool nonBlock, unsigned long tail,
29                           unsigned int logMask, pid_t pid,
30                           uint64_t start, uint64_t timeout) :
31        mRefCount(1),
32        mRelease(false),
33        mError(false),
34        threadRunning(false),
35        leadingDropped(false),
36        mReader(reader),
37        mLogMask(logMask),
38        mPid(pid),
39        mCount(0),
40        mTail(tail),
41        mIndex(0),
42        mClient(client),
43        mStart(start),
44        mNonBlock(nonBlock),
45        mEnd(LogBufferElement::getCurrentSequence()) {
46    mTimeout.tv_sec = timeout / NS_PER_SEC;
47    mTimeout.tv_nsec = timeout % NS_PER_SEC;
48    pthread_cond_init(&threadTriggeredCondition, NULL);
49    cleanSkip_Locked();
50}
51
52void LogTimeEntry::startReader_Locked(void) {
53    pthread_attr_t attr;
54
55    threadRunning = true;
56
57    if (!pthread_attr_init(&attr)) {
58        if (!pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED)) {
59            if (!pthread_create(&mThread, &attr,
60                                LogTimeEntry::threadStart, this)) {
61                pthread_attr_destroy(&attr);
62                return;
63            }
64        }
65        pthread_attr_destroy(&attr);
66    }
67    threadRunning = false;
68    if (mClient) {
69        mClient->decRef();
70    }
71    decRef_Locked();
72}
73
74void LogTimeEntry::threadStop(void *obj) {
75    LogTimeEntry *me = reinterpret_cast<LogTimeEntry *>(obj);
76
77    lock();
78
79    if (me->mNonBlock) {
80        me->error_Locked();
81    }
82
83    SocketClient *client = me->mClient;
84
85    if (me->isError_Locked()) {
86        LogReader &reader = me->mReader;
87        LastLogTimes &times = reader.logbuf().mTimes;
88
89        LastLogTimes::iterator it = times.begin();
90        while(it != times.end()) {
91            if (*it == me) {
92                times.erase(it);
93                me->release_nodelete_Locked();
94                break;
95            }
96            it++;
97        }
98
99        me->mClient = NULL;
100        reader.release(client);
101    }
102
103    if (client) {
104        client->decRef();
105    }
106
107    me->threadRunning = false;
108    me->decRef_Locked();
109
110    unlock();
111}
112
113void *LogTimeEntry::threadStart(void *obj) {
114    prctl(PR_SET_NAME, "logd.reader.per");
115
116    LogTimeEntry *me = reinterpret_cast<LogTimeEntry *>(obj);
117
118    pthread_cleanup_push(threadStop, obj);
119
120    SocketClient *client = me->mClient;
121    if (!client) {
122        me->error();
123        return NULL;
124    }
125
126    LogBuffer &logbuf = me->mReader.logbuf();
127
128    bool privileged = FlushCommand::hasReadLogs(client);
129    bool security = FlushCommand::hasSecurityLogs(client);
130
131    me->leadingDropped = true;
132
133    lock();
134
135    uint64_t start = me->mStart;
136
137    while (me->threadRunning && !me->isError_Locked()) {
138
139        if (me->mTimeout.tv_sec || me->mTimeout.tv_nsec) {
140            if (pthread_cond_timedwait(&me->threadTriggeredCondition,
141                                       &timesLock,
142                                       &me->mTimeout) == ETIMEDOUT) {
143                me->mTimeout.tv_sec = 0;
144                me->mTimeout.tv_nsec = 0;
145            }
146            if (!me->threadRunning || me->isError_Locked()) {
147                break;
148            }
149        }
150
151        unlock();
152
153        if (me->mTail) {
154            logbuf.flushTo(client, start, privileged, security, FilterFirstPass, me);
155            me->leadingDropped = true;
156        }
157        start = logbuf.flushTo(client, start, privileged, security, FilterSecondPass, me);
158
159        lock();
160
161        if (start == LogBufferElement::FLUSH_ERROR) {
162            me->error_Locked();
163            break;
164        }
165
166        me->mStart = start + 1;
167
168        if (me->mNonBlock || !me->threadRunning || me->isError_Locked()) {
169            break;
170        }
171
172        me->cleanSkip_Locked();
173
174        if (!me->mTimeout.tv_sec && !me->mTimeout.tv_nsec) {
175            pthread_cond_wait(&me->threadTriggeredCondition, &timesLock);
176        }
177    }
178
179    unlock();
180
181    pthread_cleanup_pop(true);
182
183    return NULL;
184}
185
186// A first pass to count the number of elements
187int LogTimeEntry::FilterFirstPass(const LogBufferElement *element, void *obj) {
188    LogTimeEntry *me = reinterpret_cast<LogTimeEntry *>(obj);
189
190    LogTimeEntry::lock();
191
192    if (me->leadingDropped) {
193        if (element->getDropped()) {
194            LogTimeEntry::unlock();
195            return false;
196        }
197        me->leadingDropped = false;
198    }
199
200    if (me->mCount == 0) {
201        me->mStart = element->getSequence();
202    }
203
204    if ((!me->mPid || (me->mPid == element->getPid()))
205            && (me->isWatching(element->getLogId()))) {
206        ++me->mCount;
207    }
208
209    LogTimeEntry::unlock();
210
211    return false;
212}
213
214// A second pass to send the selected elements
215int LogTimeEntry::FilterSecondPass(const LogBufferElement *element, void *obj) {
216    LogTimeEntry *me = reinterpret_cast<LogTimeEntry *>(obj);
217
218    LogTimeEntry::lock();
219
220    me->mStart = element->getSequence();
221
222    if (me->skipAhead[element->getLogId()]) {
223        me->skipAhead[element->getLogId()]--;
224        goto skip;
225    }
226
227    if (me->leadingDropped) {
228        if (element->getDropped()) {
229            goto skip;
230        }
231        me->leadingDropped = false;
232    }
233
234    // Truncate to close race between first and second pass
235    if (me->mNonBlock && me->mTail && (me->mIndex >= me->mCount)) {
236        goto stop;
237    }
238
239    if (!me->isWatching(element->getLogId())) {
240        goto skip;
241    }
242
243    if (me->mPid && (me->mPid != element->getPid())) {
244        goto skip;
245    }
246
247    if (me->isError_Locked()) {
248        goto stop;
249    }
250
251    if (!me->mTail) {
252        goto ok;
253    }
254
255    ++me->mIndex;
256
257    if ((me->mCount > me->mTail) && (me->mIndex <= (me->mCount - me->mTail))) {
258        goto skip;
259    }
260
261    if (!me->mNonBlock) {
262        me->mTail = 0;
263    }
264
265ok:
266    if (!me->skipAhead[element->getLogId()]) {
267        LogTimeEntry::unlock();
268        return true;
269    }
270    // FALLTHRU
271
272skip:
273    LogTimeEntry::unlock();
274    return false;
275
276stop:
277    LogTimeEntry::unlock();
278    return -1;
279}
280
281void LogTimeEntry::cleanSkip_Locked(void) {
282    for (log_id_t i = LOG_ID_MIN; i < LOG_ID_MAX; i = (log_id_t) (i + 1)) {
283        skipAhead[i] = 0;
284    }
285}
286