comphelper/source/misc/asyncnotification.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 <comphelper/asyncnotification.hxx>
  21 #include <comphelper/scopeguard.hxx>
  22 #include <mutex>
  23 #include <condition_variable>
  24
  25 #include <cassert>
  26 #include <stdexcept>
  27 #include <vector>
  28 #include <algorithm>
  29
  30 namespace comphelper
  31 {
  32     AnyEvent::AnyEvent()
  33     {
  34     }
  35
  36     AnyEvent::~AnyEvent()
  37     {
  38     }
  39
  40     namespace {
  41
  42     struct ProcessableEvent
  43     {
  44         AnyEventRef                         aEvent;
  45         ::rtl::Reference< IEventProcessor > xProcessor;
  46     };
  47
  48     struct EqualProcessor
  49     {
  50         const ::rtl::Reference< IEventProcessor >&  rProcessor;
  51         explicit EqualProcessor( const ::rtl::Reference< IEventProcessor >& _rProcessor ) :rProcessor( _rProcessor ) { }
  52
  53         bool operator()( const ProcessableEvent& _rEvent )
  54         {
  55             return _rEvent.xProcessor.get() == rProcessor.get();
  56         }
  57     };
  58
  59     }
  60
  61     struct EventNotifierImpl
  62     {
  63         std::mutex        aMutex;
  64         std::condition_variable aPendingActions;
  65         std::vector< ProcessableEvent > aEvents;
  66         bool                bTerminate;
  67         // only used for AsyncEventNotifierAutoJoin
  68         char const*         name;
  69         std::shared_ptr<AsyncEventNotifierAutoJoin> pKeepThisAlive;
  70
  71         EventNotifierImpl()
  72             : bTerminate(false)
  73             , name(nullptr)
  74         {
  75         }
  76     };
  77
  78     AsyncEventNotifierBase::AsyncEventNotifierBase()
  79         : m_xImpl(new EventNotifierImpl)
  80     {
  81     }
  82
  83
  84     AsyncEventNotifierBase::~AsyncEventNotifierBase()
  85     {
  86     }
  87
  88
  89     void AsyncEventNotifierBase::removeEventsForProcessor( const ::rtl::Reference< IEventProcessor >& _xProcessor )
  90     {
  91         std::scoped_lock aGuard( m_xImpl->aMutex );
  92
  93         // remove all events for this processor
  94         m_xImpl->aEvents.erase(std::remove_if( m_xImpl->aEvents.begin(), m_xImpl->aEvents.end(), EqualProcessor( _xProcessor ) ), m_xImpl->aEvents.end());
  95     }
  96
  97
  98     void SAL_CALL AsyncEventNotifierBase::terminate()
  99     {
 100         std::scoped_lock aGuard( m_xImpl->aMutex );
 101
 102         // remember the termination request
 103         m_xImpl->bTerminate = true;
 104
 105         // awake the thread
 106         m_xImpl->aPendingActions.notify_all();
 107     }
 108
 109
 110     void AsyncEventNotifierBase::addEvent( const AnyEventRef& _rEvent, const ::rtl::Reference< IEventProcessor >& _xProcessor )
 111     {
 112         std::scoped_lock aGuard( m_xImpl->aMutex );
 113
 114         // remember this event
 115         m_xImpl->aEvents.emplace_back( ProcessableEvent {_rEvent, _xProcessor} );
 116
 117         // awake the thread
 118         m_xImpl->aPendingActions.notify_all();
 119     }
 120
 121
 122     void AsyncEventNotifierBase::execute()
 123     {
 124         for (;;)
 125         {
 126             std::vector< ProcessableEvent > aEvents;
 127             {
 128                 std::unique_lock aGuard(m_xImpl->aMutex);
 129                 m_xImpl->aPendingActions.wait(aGuard,
 130                     [this] { return m_xImpl->bTerminate || !m_xImpl->aEvents.empty(); } );
 131                 if (m_xImpl->bTerminate)
 132                     return;
 133                 else
 134                     std::swap(aEvents, m_xImpl->aEvents);
 135             }
 136             for (ProcessableEvent& rEvent : aEvents)
 137             {
 138                 assert(rEvent.xProcessor.is());
 139                 rEvent.xProcessor->processEvent(*rEvent.aEvent);
 140             }
 141             aEvents.clear();
 142         }
 143     }
 144
 145     AsyncEventNotifier::AsyncEventNotifier(char const* name)
 146         : salhelper::Thread(name)
 147     {
 148     }
 149
 150     AsyncEventNotifier::~AsyncEventNotifier()
 151     {
 152     }
 153
 154     void AsyncEventNotifier::execute()
 155     {
 156         return AsyncEventNotifierBase::execute();
 157     }
 158
 159     void AsyncEventNotifier::terminate()
 160     {
 161         return AsyncEventNotifierBase::terminate();
 162     }
 163
 164     namespace {
 165
 166     std::mutex& GetTheNotifiersMutex()
 167     {
 168         static std::mutex MUTEX;
 169         return MUTEX;
 170     }
 171
 172     }
 173
 174     static std::vector<std::weak_ptr<AsyncEventNotifierAutoJoin>> g_Notifiers;
 175
 176     void JoinAsyncEventNotifiers()
 177     {
 178         std::vector<std::weak_ptr<AsyncEventNotifierAutoJoin>> notifiers;
 179         {
 180             std::scoped_lock g(GetTheNotifiersMutex());
 181             notifiers = g_Notifiers;
 182         }
 183         for (std::weak_ptr<AsyncEventNotifierAutoJoin> const& wNotifier : notifiers)
 184         {
 185             std::shared_ptr<AsyncEventNotifierAutoJoin> const pNotifier(
 186                     wNotifier.lock());
 187             if (pNotifier)
 188             {
 189                 pNotifier->terminate();
 190                 pNotifier->join();
 191             }
 192         }
 193         // note it's possible that g_Notifiers isn't empty now in case of leaks,
 194         // particularly since the UNO service manager isn't disposed yet
 195     }
 196
 197     AsyncEventNotifierAutoJoin::AsyncEventNotifierAutoJoin(char const* name)
 198     {
 199         m_xImpl->name = name;
 200     }
 201
 202     AsyncEventNotifierAutoJoin::~AsyncEventNotifierAutoJoin()
 203     {
 204         std::scoped_lock g(GetTheNotifiersMutex());
 205         // note: this doesn't happen atomically with the refcount
 206         // hence it's possible this deletes > 1 or 0 elements
 207         g_Notifiers.erase(
 208             std::remove_if(g_Notifiers.begin(), g_Notifiers.end(),
 209                 [](std::weak_ptr<AsyncEventNotifierAutoJoin> const& w) {
 210                     return w.expired();
 211                 } ),
 212             g_Notifiers.end());
 213     }
 214
 215     std::shared_ptr<AsyncEventNotifierAutoJoin>
 216     AsyncEventNotifierAutoJoin::newAsyncEventNotifierAutoJoin(char const* name)
 217     {
 218         std::shared_ptr<AsyncEventNotifierAutoJoin> const ret(
 219                 new AsyncEventNotifierAutoJoin(name));
 220         std::scoped_lock g(GetTheNotifiersMutex());
 221         g_Notifiers.push_back(ret);
 222         return ret;
 223     }
 224
 225     void AsyncEventNotifierAutoJoin::terminate()
 226     {
 227         return AsyncEventNotifierBase::terminate();
 228     }
 229
 230     void AsyncEventNotifierAutoJoin::launch(std::shared_ptr<AsyncEventNotifierAutoJoin> const& xThis)
 231     {
 232         // see salhelper::Thread::launch
 233         xThis->m_xImpl->pKeepThisAlive = xThis;
 234         comphelper::ScopeGuard g([&xThis] { xThis->m_xImpl->pKeepThisAlive.reset(); });
 235         if (!xThis->create()) {
 236             throw std::runtime_error("osl::Thread::create failed");
 237         }
 238         g.dismiss();
 239     }
 240
 241     void AsyncEventNotifierAutoJoin::run()
 242     {
 243         // see salhelper::Thread::run
 244         comphelper::ScopeGuard g([this] { onTerminated(); });
 245         setName(m_xImpl->name);
 246         execute();
 247         g.dismiss();
 248     }
 249
 250     void AsyncEventNotifierAutoJoin::onTerminated()
 251     {
 252         // try to delete "this"
 253         m_xImpl->pKeepThisAlive.reset();
 254     }
 255
 256 } // namespace comphelper
 257
 258
 259 /* vim:set shiftwidth=4 softtabstop=4 expandtab: */