dbuffer.hpp

/* SPDX-License-Identifier: MPL-2.0 */

#ifndef __ZMQ_DBUFFER_HPP_INCLUDED__
#define __ZMQ_DBUFFER_HPP_INCLUDED__

#include <stdlib.h>
#include <stddef.h>
#include <algorithm>

#include "mutex.hpp"
#include "msg.hpp"

namespace zmq
{
//  dbuffer is a single-producer single-consumer double-buffer
//  implementation.
//
//  The producer writes to a back buffer and then tries to swap
//  pointers between the back and front buffers. If it fails,
//  due to the consumer reading from the front buffer, it just
//  gives up, which is ok since writes are many and redundant.
//
//  The reader simply reads from the front buffer.
//
//  has_msg keeps track of whether there has been a not yet read
//  value written, it is used by ypipe_conflate to mimic ypipe
//  functionality regarding a reader being asleep

template <typename T> class dbuffer_t;

template <> class dbuffer_t<msg_t>
{
  public:
    dbuffer_t () : _back (&_storage[0]), _front (&_storage[1]), _has_msg (false)
    {
        _back->init ();
        _front->init ();
    }

    ~dbuffer_t ()
    {
        _back->close ();
        _front->close ();
    }

    void write (const msg_t &value_)
    {
        zmq_assert (value_.check ());
        *_back = value_;

        zmq_assert (_back->check ());

        if (_sync.try_lock ()) {
            _front->move (*_back);
            _has_msg = true;

            _sync.unlock ();
        }
    }

    bool read (msg_t *value_)
    {
        if (!value_)
            return false;

        {
            scoped_lock_t lock (_sync);
            if (!_has_msg)
                return false;

            zmq_assert (_front->check ());

            *value_ = *_front;
            _front->init (); // avoid double free

            _has_msg = false;
            return true;
        }
    }


    bool check_read ()
    {
        scoped_lock_t lock (_sync);

        return _has_msg;
    }

    bool probe (bool (*fn_) (const msg_t &))
    {
        scoped_lock_t lock (_sync);
        return (*fn_) (*_front);
    }


  private:
    msg_t _storage[2];
    msg_t *_back, *_front;

    mutex_t _sync;
    bool _has_msg;

    ZMQ_NON_COPYABLE_NOR_MOVABLE (dbuffer_t)
};
}

#endif