syncthrd.h

Go to the documentation of this file.

/*
 * syncthrd.h
 *
 * Various thread synchronisation classes.
 *
 * Portable Tools Library
 *
 * Copyright (c) 1993-1998 Equivalence Pty. Ltd.
 *
 * The contents of this file are subject to the Mozilla Public License
 * Version 1.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.mozilla.org/MPL/
 *
 * Software distributed under the License is distributed on an "AS IS"
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
 * the License for the specific language governing rights and limitations
 * under the License.
 *
 * The Original Code is Portable Windows Library.
 *
 * The Initial Developer of the Original Code is Equivalence Pty. Ltd.
 *
 * Portions are Copyright (C) 1993 Free Software Foundation, Inc.
 * All Rights Reserved.
 *
 * Contributor(s): ______________________________________.
 */

#ifndef PTLIB_SYNCTHRD_H
#define PTLIB_SYNCTHRD_H

#ifdef P_USE_PRAGMA
#pragma interface
#endif

#include <ptlib/mutex.h>
#include <ptlib/semaphor.h>
#include <ptlib/syncpoint.h>
#include <map>
#include <queue>


class PSyncPointAck : public PSyncPoint
{
  PCLASSINFO(PSyncPointAck, PSyncPoint);

  public:
    virtual void Signal();
    void Signal(const PTimeInterval & waitTime);

    void Acknowledge();

  protected:
    PSyncPoint ack;
};


class PCondMutex : public PMutex
{
  PCLASSINFO(PCondMutex, PMutex);

  public:
    virtual void WaitCondition();

    virtual void Signal();

    virtual PBoolean Condition() = 0;

    virtual void OnWait();

  protected:
    PSyncPoint syncPoint;
};


class PIntCondMutex : public PCondMutex
{
  PCLASSINFO(PIntCondMutex, PCondMutex);

  public:
    enum Operation {
      LT,
      LE,
      EQ,
      GE,
      GT
    };

    PIntCondMutex(
      int value = 0,            
      int target = 0,           
      Operation operation = LE  
    );

    void PrintOn(ostream & strm) const;

    virtual PBoolean Condition();

    operator int() const { return value; }

    PIntCondMutex & operator=(int newval);

    PIntCondMutex & operator++();

    PIntCondMutex & operator+=(int inc);

    PIntCondMutex & operator--();

    PIntCondMutex & operator-=(int dec);


  protected:
    int value, target;
    Operation operation;
};


class PReadWriteMutex : public PObject, public PMutexExcessiveLockInfo, PProfiling::HighWaterMark<PReadWriteMutex>
{
  PCLASSINFO(PReadWriteMutex, PObject);
  public:
    explicit PReadWriteMutex();
    explicit PReadWriteMutex(
      const PDebugLocation & location, 
      unsigned timeout = 0             
    );
    ~PReadWriteMutex();

    __inline void StartRead() { InternalStartRead(NULL); }
    __inline void StartRead(const PDebugLocation & location) { InternalStartRead(&location); }

    __inline void EndRead() { InternalEndRead(NULL); }
    __inline void EndRead(const PDebugLocation & location) { InternalEndRead(&location); }

    __inline void StartWrite() { InternalStartWrite(NULL); }
    __inline void StartWrite(const PDebugLocation & location) { InternalStartWrite(&location); }

    void EndWrite() { InternalEndWrite(NULL); }
    void EndWrite(const PDebugLocation & location) { InternalEndWrite(&location); }

    virtual void PrintOn(ostream &strm) const;

  protected:
    void InternalStartRead(const PDebugLocation * location);
    void InternalEndRead(const PDebugLocation * location);
    void InternalStartWrite(const PDebugLocation * location);
    void InternalEndWrite(const PDebugLocation * location);

#if P_READ_WRITE_ALGO2
    PSemaphore  m_inSemaphore;
    unsigned    m_inCount;
    PSemaphore  m_outSemaphore;
    unsigned    m_outCount;
    PSemaphore  m_writeSemaphore;
    bool        m_wait;
#else
    PSemaphore  m_readerSemaphore;
    PTimedMutex m_readerMutex;
    unsigned    m_readerCount;
    PTimedMutex m_starvationPreventer;

    PSemaphore  m_writerSemaphore;
    PTimedMutex m_writerMutex;
    unsigned    m_writerCount;
#endif
    struct Nest
    {
      atomic<unsigned> m_readerCount;
      atomic<unsigned> m_writerCount;
      atomic<bool>     m_waiting;
      atomic<uint64_t> m_startHeldCycle;
      atomic<PUniqueThreadIdentifier> m_uniqueId;

      Nest();
      Nest(const Nest & other);
      Nest & operator=(const Nest & other);
    };
    typedef std::map<PThreadIdentifier, Nest> NestMap;
    NestMap          m_nestedThreads;
    PCriticalSection m_nestingMutex;

    Nest * GetNest();
    Nest & StartNest();
    void EndNest();
    void InternalStartReadWithNest(Nest & nest, const PDebugLocation & location);
    void InternalEndReadWithNest(Nest & nest, const PDebugLocation & location);
    void InternalStartWriteWithNest(Nest & nest, const PDebugLocation & location);
    void InternalEndWriteWithNest(Nest & nest, const PDebugLocation & location);
    void InternalWait(Nest & nest, PSync & sync, const PDebugLocation & location) const;

  private:
    PReadWriteMutex(const PReadWriteMutex & other) : PObject(other), m_readerCount(), m_writerCount() { }
    void operator=(const PReadWriteMutex &) { }

  friend class PSafeObject;
  friend class PReadWaitAndSignal;
  friend class PWriteWaitAndSignal;
  friend class PInstrumentedReadWaitAndSignal;
  friend class PInstrumentedWriteWaitAndSignal;
};

#define PDECLARE_READ_WRITE_MUTEX_ARG_1(var)              struct PReadWriteMutex_##var : PReadWriteMutex { PReadWriteMutex_##var() : PReadWriteMutex(P_DEBUG_LOCATION) { } } var
#define PDECLARE_READ_WRITE_MUTEX_ARG_2(var,nam)          struct PReadWriteMutex_##var : PReadWriteMutex { PReadWriteMutex_##var() : PReadWriteMutex(#nam,           ) { } } var
#define PDECLARE_READ_WRITE_MUTEX_ARG_3(var,nam,to)       struct PReadWriteMutex_##var : PReadWriteMutex { PReadWriteMutex_##var() : PReadWriteMutex(#nam,to         ) { } } var
#define PDECLARE_READ_WRITE_MUTEX_ARG_4(var,nam,to,tw)    struct PReadWriteMutex_##var : PReadWriteMutex { PReadWriteMutex_##var() : PReadWriteMutex(#nam,to,tw      ) { } } var
#define PDECLARE_READ_WRITE_MUTEX_ARG_5(var,nam,to,tw,th) struct PReadWriteMutex_##var : PReadWriteMutex { PReadWriteMutex_##var() : PReadWriteMutex(#nam,to,tw,th   ) { } } var

#define PDECLARE_READ_WRITE_MUTEX_PART1(narg, args) PDECLARE_READ_WRITE_MUTEX_PART2(narg, args)
#define PDECLARE_READ_WRITE_MUTEX_PART2(narg, args) PDECLARE_READ_WRITE_MUTEX_ARG_##narg args

#define PDECLARE_READ_WRITE_MUTEX(...) PDECLARE_READ_WRITE_MUTEX_PART1(PARG_COUNT(__VA_ARGS__), (__VA_ARGS__))

class PReadWriteWaitAndSignalBase
{
  protected:
    typedef void (PReadWriteMutex:: * StartFn)(const PDebugLocation * location);
    typedef void (PReadWriteMutex:: * EndFn)(const PDebugLocation * location);

    PReadWriteWaitAndSignalBase(const PReadWriteMutex & mutex, const PDebugLocation * location, StartFn start, EndFn end)
      : m_mutex(const_cast<PReadWriteMutex &>(mutex))
      , m_location(location)
      , m_end(end)
    {
      if (start)
        (m_mutex.*start)(&m_location);
    }

  public:
    ~PReadWriteWaitAndSignalBase()
    {
      (m_mutex.*m_end)(&m_location);
    }

  protected:
    PReadWriteMutex & m_mutex;
    PDebugLocation    m_location;
    EndFn             m_end;
};



class PReadWaitAndSignal : public PReadWriteWaitAndSignalBase
{
  public:
    PReadWaitAndSignal(
      const PReadWriteMutex & mutex,  
      bool start = true               
    ) : PReadWriteWaitAndSignalBase(mutex, NULL, start ? &PReadWriteMutex::InternalStartRead : NULL, &PReadWriteMutex::InternalEndRead) { }
};


class PWriteWaitAndSignal : public PReadWriteWaitAndSignalBase
{
  public:
    PWriteWaitAndSignal(
      const PReadWriteMutex & mutex,  
      PBoolean start = true           
    ) : PReadWriteWaitAndSignalBase(mutex, NULL, start ? &PReadWriteMutex::InternalStartWrite : NULL, &PReadWriteMutex::InternalEndWrite) { }
};


#if PTRACING
  class PInstrumentedReadWriteMutex : public PReadWriteMutex
  {
    public:
      PInstrumentedReadWriteMutex(
        const char * baseName,
        const char * file,
        unsigned line,
        const char * waitReadOnlyName,
        const char * heldReadOnlyName,
        const char * waitReadWriteName,
        const char * heldReadWriteName,
        unsigned waitTime,
        unsigned heldTime,
        unsigned throttleTime = 10000,    
        unsigned throttledLogLevel = 2,   
        unsigned unthrottledLogLevel = 6, 
        unsigned thresholdPercent = 5,    
        unsigned maxHistory = 0           
      ) : PReadWriteMutex           (PDebugLocation(file, line, baseName         ), MinDeadlockTime(waitTime))
        , m_timeWaitReadOnlyContext (PDebugLocation(file, line, waitReadOnlyName ), waitTime, throttleTime, throttledLogLevel, unthrottledLogLevel, thresholdPercent, maxHistory)
        , m_timeHeldReadOnlyContext (PDebugLocation(file, line, heldReadOnlyName ), heldTime, throttleTime, throttledLogLevel, unthrottledLogLevel, thresholdPercent, maxHistory)
        , m_timeWaitReadWriteContext(PDebugLocation(file, line, waitReadWriteName), waitTime, throttleTime, throttledLogLevel, unthrottledLogLevel, thresholdPercent, maxHistory)
        , m_timeHeldReadWriteContext(PDebugLocation(file, line, heldReadWriteName), heldTime, throttleTime, throttledLogLevel, unthrottledLogLevel, thresholdPercent, maxHistory)
      { }

      void SetWaitReadOnlyThrottleTime(unsigned throttleTime) { m_timeWaitReadOnlyContext.SetThrottleTime(throttleTime); }
      void SetWaitReadOnlyThrottledLogLevel(unsigned throttledLogLevel) { m_timeWaitReadOnlyContext.SetThrottledLogLevel(throttledLogLevel); }
      void SetWaitReadOnlyUnthrottledLogLevel(unsigned unthrottledLogLevel) { m_timeWaitReadOnlyContext.SetUnthrottledLogLevel(unthrottledLogLevel); }
      void SetWaitReadOnlyThresholdPercent(unsigned thresholdPercent) { m_timeWaitReadOnlyContext.SetThresholdPercent(thresholdPercent); }
      void SetWaitReadOnlyMaxHistory(unsigned maxHistory) { m_timeWaitReadOnlyContext.SetMaxHistory(maxHistory); }

      void SetHeldReadOnlyThrottleTime(unsigned throttleTime) { m_timeHeldReadOnlyContext.SetThrottleTime(throttleTime); }
      void SetHeldReadOnlyThrottledLogLevel(unsigned throttledLogLevel) { m_timeHeldReadOnlyContext.SetThrottledLogLevel(throttledLogLevel); }
      void SetHeldReadOnlyUnthrottledLogLevel(unsigned unthrottledLogLevel) { m_timeHeldReadOnlyContext.SetUnthrottledLogLevel(unthrottledLogLevel); }
      void SetHeldReadOnlyThresholdPercent(unsigned thresholdPercent) { m_timeHeldReadOnlyContext.SetThresholdPercent(thresholdPercent); }
      void SetHeldReadOnlyMaxHistory(unsigned maxHistory) { m_timeHeldReadOnlyContext.SetMaxHistory(maxHistory); }

      void SetWaitReadWriteThrottleTime(unsigned throttleTime) { m_timeWaitReadWriteContext.SetThrottleTime(throttleTime); }
      void SetWaitReadWriteThrottledLogLevel(unsigned throttledLogLevel) { m_timeWaitReadWriteContext.SetThrottledLogLevel(throttledLogLevel); }
      void SetWaitReadWriteUnthrottledLogLevel(unsigned unthrottledLogLevel) { m_timeWaitReadWriteContext.SetUnthrottledLogLevel(unthrottledLogLevel); }
      void SetWaitReadWriteThresholdPercent(unsigned thresholdPercent) { m_timeWaitReadWriteContext.SetThresholdPercent(thresholdPercent); }
      void SetWaitReadWriteMaxHistory(unsigned maxHistory) { m_timeWaitReadWriteContext.SetMaxHistory(maxHistory); }

      void SetHeldReadWriteThrottleTime(unsigned throttleTime) { m_timeHeldReadWriteContext.SetThrottleTime(throttleTime); }
      void SetHeldReadWriteThrottledLogLevel(unsigned throttledLogLevel) { m_timeHeldReadWriteContext.SetThrottledLogLevel(throttledLogLevel); }
      void SetHeldReadWriteUnthrottledLogLevel(unsigned unthrottledLogLevel) { m_timeHeldReadWriteContext.SetUnthrottledLogLevel(unthrottledLogLevel); }
      void SetHeldReadWriteThresholdPercent(unsigned thresholdPercent) { m_timeHeldReadWriteContext.SetThresholdPercent(thresholdPercent); }
      void SetHeldReadWriteMaxHistory(unsigned maxHistory) { m_timeHeldReadWriteContext.SetMaxHistory(maxHistory); }

    protected:
      virtual void AcquiredLock(uint64_t startWaitCycle, bool readOnly, const PDebugLocation & location);
      virtual void ReleasedLock(const PObject & mutex, uint64_t startHeldSamplePoint, bool readOnly, const PDebugLocation & location);

      PProfiling::TimeScope m_timeWaitReadOnlyContext;
      PProfiling::TimeScope m_timeHeldReadOnlyContext;
      PProfiling::TimeScope m_timeWaitReadWriteContext;
      PProfiling::TimeScope m_timeHeldReadWriteContext;
  };

  class PInstrumentedReadWaitAndSignal : public PReadWriteWaitAndSignalBase
  {
    public:
      PInstrumentedReadWaitAndSignal(
        const PReadWriteMutex & mutex,
        const PDebugLocation & location,
        bool start = true
      ) : PReadWriteWaitAndSignalBase(mutex, &location, start ? &PReadWriteMutex::InternalStartRead : NULL, &PReadWriteMutex::InternalEndRead) { }
  };

  class PInstrumentedWriteWaitAndSignal : public PReadWriteWaitAndSignalBase
  {
    public:
      PInstrumentedWriteWaitAndSignal(
        const PReadWriteMutex & mutex,
        const PDebugLocation & location,
        PBoolean start = true
      ) : PReadWriteWaitAndSignalBase(mutex, &location, start ? &PReadWriteMutex::InternalStartWrite : NULL, &PReadWriteMutex::InternalEndWrite) { }
  };

  #define PDECLARE_INSTRUMENTED_READ_WRITE_MUTEX(var, name, ...)                \
    struct PInstrumentedReadWriteMutex_##name : PInstrumentedReadWriteMutex {   \
      PInstrumentedReadWriteMutex_##name(const char * mutexName = #name)        \
            : PInstrumentedReadWriteMutex(mutexName, __FILE__, __LINE__,        \
                                          "Wait R/O " #name, "Held R/O " #name, \
                                          "Wait R/W " #name, "Held R/W " #name, \
                                          __VA_ARGS__) { }                      \
    } var
#else
  #define PDECLARE_INSTRUMENTED_READ_WRITE_MUTEX(var, name, waitTime, heldTime, ...) \
                       PDECLARE_READ_WRITE_MUTEX(var, name, MinDeadlockTime(waitTime))
#endif


template <class T> class PSyncQueue : public PObject
{
    PCLASSINFO(PSyncQueue, PObject);
  public:
    typedef std::queue<T> BaseQueue;

    enum State
    {
      e_Open,
      e_Blocked,
      e_Draining,
      e_Closed
    };

    PSyncQueue()
      : m_state(e_Open)
      , m_available(0, INT_MAX)
    {
    }

    ~PSyncQueue()
    {
      Close(true);
    }

    bool Enqueue(const T & obj)
    {
      PWaitAndSignal lock(m_mutex);
      if (m_state == e_Closed || m_state == e_Draining)
        return false;
      m_queue.push(obj);
      m_available.Signal();
      return true;
    }

    bool Dequeue(T & value, const PTimeInterval & timeout = PMaxTimeInterval)
    {
      bool dequeued = false;

      m_mutex.Wait();

      switch (m_state) {
        case e_Blocked :
          PAssertAlways("Multiple threads in PSyncQueue::Dequeue()");
          break;

        case e_Open :
          m_state = e_Blocked;

          {
            m_mutex.Signal();
            bool available = m_available.Wait(timeout);
            m_mutex.Wait();

            if (available && !m_queue.empty()) {
              value = m_queue.front();
              m_queue.pop();
              dequeued = true;
            }
          }

          if (m_state == e_Blocked) {
            m_state = e_Open;
            break;
          }
          if (m_state == e_Draining && m_queue.empty())
            m_state = e_Closed;     // Just popped the last item
          if (m_state == e_Closed)
            m_closed.Signal();
          break;

        case e_Draining:
          if (!m_queue.empty()) {
            // Continue draining
            value = m_queue.front();
            m_queue.pop();
            dequeued = true;
            break;  
          }
          m_state = e_Closed;
          // Do closed case

        case e_Closed :
          m_closed.Signal();
      }

      m_mutex.Signal();

      return dequeued;
    }

    void Drain(bool wait)
    {
      bool blocked;
      {
        PWaitAndSignal mutex(m_mutex);
        if (m_state == e_Closed || m_state == e_Draining)
          return;

        blocked = m_state == e_Blocked;
        m_state = m_queue.empty() ? e_Closed : e_Draining;
        m_available.Signal();
      }

      if (blocked && wait)
        m_closed.Wait();
    }

    void Close(bool wait)
    {
      m_mutex.Wait();

      bool blocked = m_state == e_Blocked;
      m_state = e_Closed;

      while (!m_queue.empty())
        m_queue.pop();

      m_available.Signal();
      m_mutex.Signal();

      if (blocked && wait)
        m_closed.Wait();
    }

    // Indicate queue is in use.
    bool IsOpen() const
    {
        PWaitAndSignal mutex(m_mutex);
        return m_state != e_Closed;
    }

    void Restart()
    {
      m_mutex.Wait();
      if (m_state == e_Closed) {
        while (!m_queue.empty())
          m_queue.pop();
        while (m_available.Wait(0))
          ;
        m_state = e_Open;
      }
      m_mutex.Signal();
    }


    size_t size() const
    {
      PWaitAndSignal lock(m_mutex);
      return m_queue.size();
    }


    bool empty() const
    {
      PWaitAndSignal lock(m_mutex);
      return m_queue.empty();
    }


    __inline const PMutex & GetMutex() const { return m_mutex; }
    __inline       PMutex & GetMutex()       { return m_mutex; }

  protected:
    BaseQueue      m_queue;
    State          m_state;
    PSemaphore     m_available;
    PDECLARE_MUTEX(m_mutex);
    PSyncPoint     m_closed;

  private:
    __inline PSyncQueue(const PSyncQueue & other) : PObject(other) { }
    __inline void operator=(const PSyncQueue &) { }
};


#endif // PTLIB_SYNCTHRD_H


// End Of File ///////////////////////////////////////////////////////////////