cppu/source/threadpool/jobqueue.cxx

   1 /* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
   2 /*
   3  * This file is part of the LibreOffice project.
   4  *
   5  * This Source Code Form is subject to the terms of the Mozilla Public
   6  * License, v. 2.0. If a copy of the MPL was not distributed with this
   7  * file, You can obtain one at http://mozilla.org/MPL/2.0/.
   8  *
   9  * This file incorporates work covered by the following license notice:
  10  *
  11  *   Licensed to the Apache Software Foundation (ASF) under one or more
  12  *   contributor license agreements. See the NOTICE file distributed
  13  *   with this work for additional information regarding copyright
  14  *   ownership. The ASF licenses this file to you under the Apache
  15  *   License, Version 2.0 (the "License"); you may not use this file
  16  *   except in compliance with the License. You may obtain a copy of
  17  *   the License at http://www.apache.org/licenses/LICENSE-2.0 .
  18  */
  19
  20 #include <sal/config.h>
  21
  22 #include <cassert>
  23
  24 #include "jobqueue.hxx"
  25 #include "threadpool.hxx"
  26
  27 namespace cppu_threadpool {
  28
  29     JobQueue::JobQueue() :
  30         m_nToDo( 0 ),
  31         m_bSuspended( false ),
  32         m_DisposedCallerAdmin( DisposedCallerAdmin::getInstance() )
  33     {
  34     }
  35
  36     void JobQueue::add( void *pThreadSpecificData, RequestFun * doRequest )
  37     {
  38         std::scoped_lock guard( m_mutex );
  39         Job job = { pThreadSpecificData , doRequest };
  40         m_lstJob.push_back( job );
  41         if( ! m_bSuspended )
  42         {
  43             m_cndWait.notify_all();
  44         }
  45         m_nToDo ++;
  46     }
  47
  48     void *JobQueue::enter( void const * nDisposeId , bool bReturnWhenNoJob )
  49     {
  50         void *pReturn = nullptr;
  51         {
  52             // synchronize with the dispose calls
  53             std::scoped_lock guard( m_mutex );
  54             if( m_DisposedCallerAdmin->isDisposed( nDisposeId ) )
  55             {
  56                 return nullptr;
  57             }
  58             m_lstCallstack.push_front( nDisposeId );
  59         }
  60
  61
  62         while( true )
  63         {
  64             struct Job job={nullptr,nullptr};
  65             {
  66                 std::unique_lock guard( m_mutex );
  67
  68                 while (m_bSuspended
  69                        || (m_lstCallstack.front() != nullptr && !bReturnWhenNoJob
  70                            && m_lstJob.empty()))
  71                 {
  72                     m_cndWait.wait(guard);
  73                 }
  74
  75                 if( nullptr == m_lstCallstack.front() )
  76                 {
  77                     // disposed !
  78                     if (!m_lstJob.empty() && m_lstJob.front().doRequest == nullptr) {
  79                         // If this thread was waiting for a remote response, that response may or
  80                         // may not have been enqueued; if it has not been enqueued, there cannot be
  81                         // another enqueued response, so it is always correct to remove any enqueued
  82                         // response here:
  83                         m_lstJob.pop_front();
  84                     }
  85                     break;
  86                 }
  87
  88                 if( m_lstJob.empty() )
  89                 {
  90                     assert(bReturnWhenNoJob);
  91                     break;
  92                 }
  93
  94                 job = m_lstJob.front();
  95                 m_lstJob.pop_front();
  96             }
  97
  98             if( job.doRequest )
  99             {
 100                 job.doRequest( job.pThreadSpecificData );
 101                 std::scoped_lock guard( m_mutex );
 102                 m_nToDo --;
 103             }
 104             else
 105             {
 106                 pReturn = job.pThreadSpecificData;
 107                 std::scoped_lock guard( m_mutex );
 108                 m_nToDo --;
 109                 break;
 110             }
 111         }
 112
 113         {
 114             // synchronize with the dispose calls
 115             std::scoped_lock guard( m_mutex );
 116             m_lstCallstack.pop_front();
 117         }
 118
 119         return pReturn;
 120     }
 121
 122     void JobQueue::dispose( void const * nDisposeId )
 123     {
 124         std::scoped_lock guard( m_mutex );
 125         for( auto& rId : m_lstCallstack )
 126         {
 127             if( rId == nDisposeId )
 128             {
 129                 rId = nullptr;
 130             }
 131         }
 132
 133         if( !m_lstCallstack.empty()  && ! m_lstCallstack.front() )
 134         {
 135             // The thread is waiting for a disposed pCallerId, let it go
 136             m_cndWait.notify_all();
 137         }
 138     }
 139
 140     void JobQueue::suspend()
 141     {
 142         std::scoped_lock guard( m_mutex );
 143         m_bSuspended = true;
 144     }
 145
 146     void JobQueue::resume()
 147     {
 148         std::scoped_lock guard( m_mutex );
 149         m_bSuspended = false;
 150         if( ! m_lstJob.empty() )
 151         {
 152             m_cndWait.notify_all();
 153         }
 154     }
 155
 156     bool JobQueue::isEmpty() const
 157     {
 158         std::scoped_lock guard( m_mutex );
 159         return m_lstJob.empty();
 160     }
 161
 162     bool JobQueue::isCallstackEmpty() const
 163     {
 164         std::scoped_lock guard( m_mutex );
 165         return m_lstCallstack.empty();
 166     }
 167
 168     bool JobQueue::isBusy() const
 169     {
 170         std::scoped_lock guard( m_mutex );
 171         return m_nToDo > 0;
 172     }
 173
 174
 175 }
 176
 177 /* vim:set shiftwidth=4 softtabstop=4 expandtab: */