Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit 4f22786c authored by Tom Cherry's avatar Tom Cherry
Browse files

logd: rework logic for LogTimeEntry 

LogTimeEntry's lifecycle is spread out in various locations.  It
further seems incomplete as there is logic that assumes that its
associated thread can exit while the underlying LogTimeEntry remains
valid, however it doesn't appear that that is actually a supported
situation.

This change simplifies this logic to have only one valid state for a
LogTimeEntry: it must have its thread running and be present in
LastLogTimes.  A LogTimeEntry will never be placed into LastLogTimes
unless its thread is running and its thread will remove its associated
LogTimeEntry from LastLogTimes before it has exited.

This admittedly breaks situations where a blocking socket gets issued
multiple commands with different pid filters, tail lines, etc,
however, I'm reasonably sure that these situations were already
broken.  A check is added to close the socket in this case.

Test: multiple logcat instances work, logd.reader.per's are cleaned up
Change-Id: Ibe8651e7d530c5e9a8d6ce3150cd247982887cbe
parent b3bc8427

logd/FlushCommand.cpp

+2 −21
Original line number Original line Diff line number Diff line
@@ -42,7 +42,7 @@ void FlushCommand::runSocketCommand(SocketClient* client) { 
    LogTimeEntry::wrlock();

    LogTimeEntry::wrlock();

    LastLogTimes::iterator it = times.begin();

    LastLogTimes::iterator it = times.begin();

    while (it != times.end()) {

    while (it != times.end()) {

        entry = (*it);

        entry = it->get();

        if (entry->mClient == client) {

        if (entry->mClient == client) {

            if (!entry->isWatchingMultiple(mLogMask)) {

            if (!entry->isWatchingMultiple(mLogMask)) {

                LogTimeEntry::unlock();

                LogTimeEntry::unlock();
@@ -63,31 +63,12 @@ void FlushCommand::runSocketCommand(SocketClient* client) { 
                }

                }

            }

            }

            entry->triggerReader_Locked();

            entry->triggerReader_Locked();

            if (entry->runningReader_Locked()) {

            LogTimeEntry::unlock();

            LogTimeEntry::unlock();

            return;

            return;

        }

        }

            entry->incRef_Locked();

            break;

        }

        it++;

        it++;

    }

    }





    if (it == times.end()) {

        // Create LogTimeEntry in notifyNewLog() ?

        if (mTail == (unsigned long)-1) {

            LogTimeEntry::unlock();

            return;

        }

        entry = new LogTimeEntry(mReader, client, mNonBlock, mTail, mLogMask,

                                 mPid, mStart, mTimeout);

        times.push_front(entry);

    }



    client->incRef();



    // release client and entry reference counts once done

    entry->startReader_Locked();

    LogTimeEntry::unlock();

    LogTimeEntry::unlock();

}

}

logd/FlushCommand.h

+1 −26
Original line number Original line Diff line number Diff line
@@ -27,36 +27,11 @@ class LogReader; 




class FlushCommand : public SocketClientCommand {

class FlushCommand : public SocketClientCommand {

    LogReader& mReader;

    LogReader& mReader;

    bool mNonBlock;

    unsigned long mTail;

    log_mask_t mLogMask;

    log_mask_t mLogMask;

    pid_t mPid;

    log_time mStart;

    uint64_t mTimeout;





   public:

   public:

    // for opening a reader

    explicit FlushCommand(LogReader& reader, bool nonBlock, unsigned long tail,

                          log_mask_t logMask, pid_t pid, log_time start,

                          uint64_t timeout)

        : mReader(reader),

          mNonBlock(nonBlock),

          mTail(tail),

          mLogMask(logMask),

          mPid(pid),

          mStart(start),

          mTimeout((start != log_time::EPOCH) ? timeout : 0) {

    }



    // for notification of an update

    explicit FlushCommand(LogReader& reader, log_mask_t logMask)

    explicit FlushCommand(LogReader& reader, log_mask_t logMask)

        : mReader(reader),

        : mReader(reader), mLogMask(logMask) {

          mNonBlock(false),

          mTail(-1),

          mLogMask(logMask),

          mPid(0),

          mStart(log_time::EPOCH),

          mTimeout(0) {

    }

    }





    virtual void runSocketCommand(SocketClient* client);

    virtual void runSocketCommand(SocketClient* client);

logd/LogBuffer.cpp

+15 −19
Original line number Original line Diff line number Diff line
@@ -105,10 +105,8 @@ void LogBuffer::init() { 




    LastLogTimes::iterator times = mTimes.begin();

    LastLogTimes::iterator times = mTimes.begin();

    while (times != mTimes.end()) {

    while (times != mTimes.end()) {

        LogTimeEntry* entry = (*times);

        LogTimeEntry* entry = times->get();

        if (entry->owned_Locked()) {

        entry->triggerReader_Locked();

        entry->triggerReader_Locked();

        }

        times++;

        times++;

    }

    }
@@ -409,8 +407,7 @@ void LogBuffer::log(LogBufferElement* elem) { 




        LastLogTimes::iterator times = mTimes.begin();

        LastLogTimes::iterator times = mTimes.begin();

        while (times != mTimes.end()) {

        while (times != mTimes.end()) {

            LogTimeEntry* entry = (*times);

            LogTimeEntry* entry = times->get();

            if (entry->owned_Locked()) {

            if (!entry->mNonBlock) {

            if (!entry->mNonBlock) {

                end_always = true;

                end_always = true;

                break;

                break;
@@ -420,7 +417,6 @@ void LogBuffer::log(LogBufferElement* elem) { 
                end = entry->mEnd;

                end = entry->mEnd;

                end_set = true;

                end_set = true;

            }

            }

            }

            times++;

            times++;

        }

        }
@@ -710,8 +706,8 @@ bool LogBuffer::prune(log_id_t id, unsigned long pruneRows, uid_t caller_uid) { 
    // Region locked?

    // Region locked?

    LastLogTimes::iterator times = mTimes.begin();

    LastLogTimes::iterator times = mTimes.begin();

    while (times != mTimes.end()) {

    while (times != mTimes.end()) {

        LogTimeEntry* entry = (*times);

        LogTimeEntry* entry = times->get();

        if (entry->owned_Locked() && entry->isWatching(id) &&

        if (entry->isWatching(id) &&

            (!oldest || (oldest->mStart > entry->mStart) ||

            (!oldest || (oldest->mStart > entry->mStart) ||

             ((oldest->mStart == entry->mStart) &&

             ((oldest->mStart == entry->mStart) &&

              (entry->mTimeout.tv_sec || entry->mTimeout.tv_nsec)))) {

              (entry->mTimeout.tv_sec || entry->mTimeout.tv_nsec)))) {
@@ -1052,9 +1048,9 @@ bool LogBuffer::clear(log_id_t id, uid_t uid) { 
                LogTimeEntry::wrlock();

                LogTimeEntry::wrlock();

                LastLogTimes::iterator times = mTimes.begin();

                LastLogTimes::iterator times = mTimes.begin();

                while (times != mTimes.end()) {

                while (times != mTimes.end()) {

                    LogTimeEntry* entry = (*times);

                    LogTimeEntry* entry = times->get();

                    // Killer punch

                    // Killer punch

                    if (entry->owned_Locked() && entry->isWatching(id)) {

                    if (entry->isWatching(id)) {

                        entry->release_Locked();

                        entry->release_Locked();

                    }

                    }

                    times++;

                    times++;

logd/LogReader.cpp

+27 −4
Original line number Original line Diff line number Diff line
@@ -41,6 +41,7 @@ void LogReader::notifyNewLog(log_mask_t logMask) { 
    runOnEachSocket(&command);

    runOnEachSocket(&command);

}

}





// Note returning false will release the SocketClient instance.

bool LogReader::onDataAvailable(SocketClient* cli) {

bool LogReader::onDataAvailable(SocketClient* cli) {

    static bool name_set;

    static bool name_set;

    if (!name_set) {

    if (!name_set) {
@@ -57,6 +58,18 @@ bool LogReader::onDataAvailable(SocketClient* cli) { 
    }

    }

    buffer[len] = '\0';

    buffer[len] = '\0';





    // Clients are only allowed to send one command, disconnect them if they

    // send another.

    LogTimeEntry::wrlock();

    for (const auto& entry : mLogbuf.mTimes) {

        if (entry->mClient == cli) {

            entry->release_Locked();

            LogTimeEntry::unlock();

            return false;

        }

    }

    LogTimeEntry::unlock();



    unsigned long tail = 0;

    unsigned long tail = 0;

    static const char _tail[] = " tail=";

    static const char _tail[] = " tail=";

    char* cp = strstr(buffer, _tail);

    char* cp = strstr(buffer, _tail);
@@ -199,14 +212,25 @@ bool LogReader::onDataAvailable(SocketClient* cli) { 
        cli->getUid(), cli->getGid(), cli->getPid(), nonBlock ? 'n' : 'b', tail,

        cli->getUid(), cli->getGid(), cli->getPid(), nonBlock ? 'n' : 'b', tail,

        logMask, (int)pid, sequence.nsec(), timeout);

        logMask, (int)pid, sequence.nsec(), timeout);





    FlushCommand command(*this, nonBlock, tail, logMask, pid, sequence, timeout);

    LogTimeEntry::wrlock();

    auto entry = std::make_unique<LogTimeEntry>(

        *this, cli, nonBlock, tail, logMask, pid, sequence, timeout);

    if (!entry->startReader_Locked()) {

        LogTimeEntry::unlock();

        return false;

    }



    // release client and entry reference counts once done

    cli->incRef();

    mLogbuf.mTimes.emplace_front(std::move(entry));





    // Set acceptable upper limit to wait for slow reader processing b/27242723

    // Set acceptable upper limit to wait for slow reader processing b/27242723

    struct timeval t = { LOGD_SNDTIMEO, 0 };

    struct timeval t = { LOGD_SNDTIMEO, 0 };

    setsockopt(cli->getSocket(), SOL_SOCKET, SO_SNDTIMEO, (const char*)&t,

    setsockopt(cli->getSocket(), SOL_SOCKET, SO_SNDTIMEO, (const char*)&t,

               sizeof(t));

               sizeof(t));





    command.runSocketCommand(cli);

    LogTimeEntry::unlock();



    return true;

    return true;

}

}
@@ -215,9 +239,8 @@ void LogReader::doSocketDelete(SocketClient* cli) { 
    LogTimeEntry::wrlock();

    LogTimeEntry::wrlock();

    LastLogTimes::iterator it = times.begin();

    LastLogTimes::iterator it = times.begin();

    while (it != times.end()) {

    while (it != times.end()) {

        LogTimeEntry* entry = (*it);

        LogTimeEntry* entry = it->get();

        if (entry->mClient == cli) {

        if (entry->mClient == cli) {

            times.erase(it);

            entry->release_Locked();

            entry->release_Locked();

            break;

            break;

        }

        }

logd/LogTimes.cpp

+21 −68
Original line number Original line Diff line number Diff line
@@ -30,11 +30,7 @@ pthread_mutex_t LogTimeEntry::timesLock = PTHREAD_MUTEX_INITIALIZER; 
LogTimeEntry::LogTimeEntry(LogReader& reader, SocketClient* client,

LogTimeEntry::LogTimeEntry(LogReader& reader, SocketClient* client,

                           bool nonBlock, unsigned long tail, log_mask_t logMask,

                           bool nonBlock, unsigned long tail, log_mask_t logMask,

                           pid_t pid, log_time start, uint64_t timeout)

                           pid_t pid, log_time start, uint64_t timeout)

    : mRefCount(1),

    : leadingDropped(false),

      mRelease(false),

      mError(false),

      threadRunning(false),

      leadingDropped(false),

      mReader(reader),

      mReader(reader),

      mLogMask(logMask),

      mLogMask(logMask),

      mPid(pid),

      mPid(pid),
@@ -52,65 +48,21 @@ LogTimeEntry::LogTimeEntry(LogReader& reader, SocketClient* client, 
    cleanSkip_Locked();

    cleanSkip_Locked();

}

}





void LogTimeEntry::startReader_Locked(void) {

bool LogTimeEntry::startReader_Locked() {

    pthread_attr_t attr;

    pthread_attr_t attr;





    threadRunning = true;



    if (!pthread_attr_init(&attr)) {

    if (!pthread_attr_init(&attr)) {

        if (!pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED)) {

        if (!pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED)) {

            if (!pthread_create(&mThread, &attr, LogTimeEntry::threadStart,

            if (!pthread_create(&mThread, &attr, LogTimeEntry::threadStart,

                                this)) {

                                this)) {

                pthread_attr_destroy(&attr);

                pthread_attr_destroy(&attr);

                return;

                return true;

            }

            }

        }

        }

        pthread_attr_destroy(&attr);

        pthread_attr_destroy(&attr);

    }

    }

    threadRunning = false;

    if (mClient) {

        mClient->decRef();

    }

    decRef_Locked();

}



void LogTimeEntry::threadStop(void* obj) {

    LogTimeEntry* me = reinterpret_cast<LogTimeEntry*>(obj);



    wrlock();



    if (me->mNonBlock) {

        me->error_Locked();

    }



    SocketClient* client = me->mClient;



    if (me->isError_Locked()) {

        LogReader& reader = me->mReader;

        LastLogTimes& times = reader.logbuf().mTimes;



        LastLogTimes::iterator it = times.begin();

        while (it != times.end()) {

            if (*it == me) {

                times.erase(it);

                me->release_nodelete_Locked();

                break;

            }

            it++;

        }



        me->mClient = nullptr;

        reader.release(client);

    }





    if (client) {

    return false;

        client->decRef();

    }



    me->threadRunning = false;

    me->decRef_Locked();



    unlock();

}

}





void* LogTimeEntry::threadStart(void* obj) {

void* LogTimeEntry::threadStart(void* obj) {
@@ -118,13 +70,7 @@ void* LogTimeEntry::threadStart(void* obj) { 




    LogTimeEntry* me = reinterpret_cast<LogTimeEntry*>(obj);

    LogTimeEntry* me = reinterpret_cast<LogTimeEntry*>(obj);





    pthread_cleanup_push(threadStop, obj);



    SocketClient* client = me->mClient;

    SocketClient* client = me->mClient;

    if (!client) {

        me->error();

        return nullptr;

    }





    LogBuffer& logbuf = me->mReader.logbuf();

    LogBuffer& logbuf = me->mReader.logbuf();
@@ -137,14 +83,14 @@ void* LogTimeEntry::threadStart(void* obj) { 




    log_time start = me->mStart;

    log_time start = me->mStart;





    while (me->threadRunning && !me->isError_Locked()) {

    while (!me->mRelease) {

        if (me->mTimeout.tv_sec || me->mTimeout.tv_nsec) {

        if (me->mTimeout.tv_sec || me->mTimeout.tv_nsec) {

            if (pthread_cond_timedwait(&me->threadTriggeredCondition,

            if (pthread_cond_timedwait(&me->threadTriggeredCondition,

                                       &timesLock, &me->mTimeout) == ETIMEDOUT) {

                                       &timesLock, &me->mTimeout) == ETIMEDOUT) {

                me->mTimeout.tv_sec = 0;

                me->mTimeout.tv_sec = 0;

                me->mTimeout.tv_nsec = 0;

                me->mTimeout.tv_nsec = 0;

            }

            }

            if (!me->threadRunning || me->isError_Locked()) {

            if (me->mRelease) {

                break;

                break;

            }

            }

        }

        }
@@ -162,13 +108,12 @@ void* LogTimeEntry::threadStart(void* obj) { 
        wrlock();

        wrlock();





        if (start == LogBufferElement::FLUSH_ERROR) {

        if (start == LogBufferElement::FLUSH_ERROR) {

            me->error_Locked();

            break;

            break;

        }

        }





        me->mStart = start + log_time(0, 1);

        me->mStart = start + log_time(0, 1);





        if (me->mNonBlock || !me->threadRunning || me->isError_Locked()) {

        if (me->mNonBlock || me->mRelease) {

            break;

            break;

        }

        }
@@ -179,9 +124,21 @@ void* LogTimeEntry::threadStart(void* obj) { 
        }

        }

    }

    }





    unlock();

    LogReader& reader = me->mReader;

    reader.release(client);





    pthread_cleanup_pop(true);

    client->decRef();



    LastLogTimes& times = reader.logbuf().mTimes;

    auto it =

        std::find_if(times.begin(), times.end(),

                     [&me](const auto& other) { return other.get() == me; });



    if (it != times.end()) {

        times.erase(it);

    }



    unlock();





    return nullptr;

    return nullptr;

}

}
@@ -247,10 +204,6 @@ int LogTimeEntry::FilterSecondPass(const LogBufferElement* element, void* obj) { 
        goto skip;

        goto skip;

    }

    }





    if (me->isError_Locked()) {

        goto stop;

    }



    if (!me->mTail) {

    if (!me->mTail) {

        goto ok;

        goto ok;

    }

    }