structbuf.h

/*
 * core/structbuf - memory or file-backed storage of structures
 *
 * Copyright (C) 2014  ARPA-SIM <urpsim@smr.arpa.emr.it>
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
 *
 * Author: Enrico Zini <enrico@enricozini.com>
 */
 
#ifndef DBALLE_CORE_STRUCTBUF_H
#define DBALLE_CORE_STRUCTBUF_H
 
#include <cstdlib>
#include <sys/mman.h>
#include <unistd.h>
#include <wreport/error.h>
 
namespace dballe {
 
namespace structbuf {
int make_anonymous_tmpfile();
void write_buffer(int fd, void* buf, size_t size);
} // namespace structbuf

template <typename T, int bufsize = 1024> class Structbuf
{
protected:
    T* membuf = nullptr;

    unsigned membuf_last = 0;

    const T* readbuf = (const T*)MAP_FAILED;

    size_t m_count = 0;

    int tmpfile_fd = -1;
 
public:
    Structbuf() : membuf(new T[bufsize]) {}
    ~Structbuf()
    {
        delete[] membuf;
        if (tmpfile_fd != -1)
        {
            if (readbuf != MAP_FAILED)
                munmap(const_cast<T*>(readbuf), m_count * sizeof(T));
            ::close(tmpfile_fd);
        }
    }

    size_t size() const { return m_count; }

    bool is_file_backed() const { return tmpfile_fd != -1; }

    void append(const T& val)
    {
        if (readbuf != MAP_FAILED)
            throw wreport::error_consistency(
                "writing to a Structbuf that is already being read");
        if (membuf_last == bufsize)
            write_to_file();
        membuf[membuf_last++] = val;
        ++m_count;
    }

    void ready_to_read()
    {
        if (tmpfile_fd == -1)
            readbuf = membuf;
        else
        {
            // Flush the remaining memory data to file
            if (membuf_last)
                write_to_file();
 
            // mmap the file for reading
            readbuf = (const T*)mmap(nullptr, m_count * sizeof(T), PROT_READ,
                                     MAP_SHARED, tmpfile_fd, 0);
            if (readbuf == MAP_FAILED)
                throw wreport::error_system(
                    "cannot map temporary file contents to memory");
        }
    }

    const T& operator[](size_t idx) const { return readbuf[idx]; }
 
protected:
    void write_to_file()
    {
        if (tmpfile_fd == -1)
            tmpfile_fd = structbuf::make_anonymous_tmpfile();
        structbuf::write_buffer(tmpfile_fd, membuf, sizeof(T) * membuf_last);
        membuf_last = 0;
    }
};
 
} // namespace dballe
 
#endif