/*
** Copyright (C) 2012 by Carnegie Mellon University.
**
** @OPENSOURCE_HEADER_START@
**
** Use of the SILK system and related source code is subject to the terms
** of the following licenses:
**
** GNU Public License (GPL) Rights pursuant to Version 2, June 1991
** Government Purpose License Rights (GPLR) pursuant to DFARS 252.227.7013
**
** NO WARRANTY
**
** ANY INFORMATION, MATERIALS, SERVICES, INTELLECTUAL PROPERTY OR OTHER
** PROPERTY OR RIGHTS GRANTED OR PROVIDED BY CARNEGIE MELLON UNIVERSITY
** PURSUANT TO THIS LICENSE (HEREINAFTER THE "DELIVERABLES") ARE ON AN
** "AS-IS" BASIS. CARNEGIE MELLON UNIVERSITY MAKES NO WARRANTIES OF ANY
** KIND, EITHER EXPRESS OR IMPLIED AS TO ANY MATTER INCLUDING, BUT NOT
** LIMITED TO, WARRANTY OF FITNESS FOR A PARTICULAR PURPOSE,
** MERCHANTABILITY, INFORMATIONAL CONTENT, NONINFRINGEMENT, OR ERROR-FREE
** OPERATION. CARNEGIE MELLON UNIVERSITY SHALL NOT BE LIABLE FOR INDIRECT,
** SPECIAL OR CONSEQUENTIAL DAMAGES, SUCH AS LOSS OF PROFITS OR INABILITY
** TO USE SAID INTELLECTUAL PROPERTY, UNDER THIS LICENSE, REGARDLESS OF
** WHETHER SUCH PARTY WAS AWARE OF THE POSSIBILITY OF SUCH DAMAGES.
** LICENSEE AGREES THAT IT WILL NOT MAKE ANY WARRANTY ON BEHALF OF
** CARNEGIE MELLON UNIVERSITY, EXPRESS OR IMPLIED, TO ANY PERSON
** CONCERNING THE APPLICATION OF OR THE RESULTS TO BE OBTAINED WITH THE
** DELIVERABLES UNDER THIS LICENSE.
**
** Licensee hereby agrees to defend, indemnify, and hold harmless Carnegie
** Mellon University, its trustees, officers, employees, and agents from
** all claims or demands made against them (and any related losses,
** expenses, or attorney's fees) arising out of, or relating to Licensee's
** and/or its sub licensees' negligent use or willful misuse of or
** negligent conduct or willful misconduct regarding the Software,
** facilities, or other rights or assistance granted by Carnegie Mellon
** University under this License, including, but not limited to, any
** claims of product liability, personal injury, death, damage to
** property, or violation of any laws or regulations.
**
** Carnegie Mellon University Software Engineering Institute authored
** documents are sponsored by the U.S. Department of Defense under
** Contract FA8721-05-C-0003. Carnegie Mellon University retains
** copyrights in all material produced under this contract. The U.S.
** Government retains a non-exclusive, royalty-free license to publish or
** reproduce these documents, or allow others to do so, for U.S.
** Government purposes only pursuant to the copyright license under the
** contract clause at 252.227.7013.
**
** @OPENSOURCE_HEADER_END@
*/

#hdr
#include <silk/silk.h>

RCSIDENTVAR(rcsID_CONN_MEMORY_H, "$Id$");

#include "connector.hh"
#end

#src
RCSIDENT("$Id$");

#include <silk/utils.h>

#ifdef SKCONNECTOR_TRACE_LEVEL
#define TRACEMSG_LEVEL SKCONNECTOR_TRACE_LEVEL
#endif
#define TRACEMSG(lvl, msg) TRACEMSG_TO_TRACEMSGLVL(lvl, msg)
#include <silk/sktracemsg.h>

/*
 *
 *    Tracing macros
 *
 */

/* Levels at which certain types of tracing are emitted */
#define FUNCTION_LEVEL 3        /* Main function tracing */
#define CHUNK_LEVEL    4        /* Chunk tracing */

/*
 *    TRACE_FUNC;
 *
 *    Emits a trace message for the containing function.  The pointer
 *    emitted is the connection object.
 */
#define TRACE_FUNC                                                      \
    TRACEMSG(FUNCTION_LEVEL, ("Function: (%p) %s", conn, __func__))


/*
 *    TRACE_MEM_CHUNK(chunk);
 *
 *    Emits trace info for a conn_mem_chunk_t data structure.
 */
#define TRACE_MEM_CHUNK(chunk)                                  \
    TRACEMSG(CHUNK_LEVEL,                                       \
             (("%s:%d Chunk: %p, writer_pos: %zd"               \
               "  reader_pos: %zd  max_reader_pos: %zd"),       \
              __FILE__, __LINE__,                               \
              chunk, chunk->writer_pos, chunk->reader_pos,      \
              chunk->max_reader_pos));


/* TYPEDEFS AND DEFINES */

/* minimum size we allow for a chunk is 4k */
#define CONN_MEM_MIN_CHUNK_SIZE     0x1000

/* standard chunk size is half a meg */
#define CONN_MEM_STD_CHUNK_SIZE     0x80000

/* standard number of chunks the connector norally has---used to
 * compute the maximum allocation */
#define CONN_MEM_STD_NUMBER_CHUNKS  3

/*
 *
 *    Macros common to all connectors
 *
 */

/* when the writer wraps, ensure this much space between the end of
 * the writer's block and the start of the reader's block */
#define WRAP_GAP  sizeof(uint64_t)

/* total space required for a block that holds 'bts_size' bytes of
 * memory in its 'data' member; includes the 'block_size' member and
 * ensures 'data' is 64-bit aligned */
#define BLOCK_TOTAL_SIZE(bts_size)                              \
    (((bts_size) + 2 * sizeof(uint64_t) - 1) & ~(UINT64_C(7)))

/* max size a caller may request; we require at least 3 blocks per
 * chunk; allow for overhead within the chunk */
#define BLOCK_MAX_SIZE_FROM_CHUNK_SIZE(bms_chunk_size)  \
    (((bms_chunk_size) - 4 * WRAP_GAP) / 3)

#end

namespace silk {

class MemoryConnector : public Connector
{
  private:
    struct conn_mem_chunk_t {
        /* pointer to the next chunk */
        conn_mem_chunk_t   *next;

        /* the data used for conn_block_t, which 'writer_pos' and
         * 'reader_pos' are offsets into */
        uint8_t            *blocks;

        /* offset to the location for the upstream writer */
        size_t              writer_pos;
        /* offset to the location for the downstream reader */
        size_t              reader_pos;

        /* total number of bytes in 'blocks'; this space includes blocks
         * the callers access and any overhead */
        size_t              capacity;

        /* last valid byte that can be read in 'blocks'; used to determine
         * when the reader_pos needs to be reset to 0 */
        size_t              max_reader_pos;
    };


  public:

    const char *get_type() const
    {
        return CONN_MEM_NAME;
    }

  private:

    static void setup_class();

    static void teardown_class(
        void *dummy);

    MemoryConnector();

    virtual ~MemoryConnector();

    conn_mem_chunk_t *chunk_alloc()
    {
        conn_mem_chunk_t *chunk;

        chunk = new conn_mem_chunk_t;
        chunk->blocks = new uint8_t[chunk_size];
        chunk->capacity = chunk_size;
        return chunk;
    }

    void chunk_free(
        conn_mem_chunk_t   *chunk)
    {
        if (chunk) {
            if (chunk->blocks) {
                delete[] chunk->blocks;
            }
            delete chunk;
        }
    }

    void chunk_pop()
    {
        conn_mem_chunk_t *chunk;

        assert(reader_chunk);
        assert(reader_chunk->next);

        if (!spare_chunk) {
            spare_chunk = reader_chunk;
            reader_chunk = reader_chunk->next;
        } else {
            chunk = reader_chunk;
            reader_chunk = reader_chunk->next;
            chunk_free(chunk);
        }
    }

    void chunk_push()
    {
        if (spare_chunk) {
            spare_chunk->next = NULL;
            spare_chunk->writer_pos = 0;
            spare_chunk->reader_pos = 0;
            spare_chunk->max_reader_pos = 0;
            writer_chunk->next = spare_chunk;
            writer_chunk = writer_chunk->next;
            spare_chunk = NULL;
        } else {
            writer_chunk->next = chunk_alloc();
            writer_chunk = writer_chunk->next;
        }
    }

    int _get_read_block(
        const conn_block_t    **reader_block,
        size_t                  size,
        bool                    no_wait)
    {
        conn_mem_chunk_t *chunk;

        assert(reader_block);
        assert(size);

        for (;;) {
            chunk = reader_chunk;
            if (chunk) {
                if (stopped) {
                    return -1;
                }

                if (chunk->reader_pos != chunk->writer_pos) {
                    if (chunk->max_reader_pos
                        && (chunk->reader_pos == chunk->max_reader_pos))
                    {
                        /* wrap the reader if the writer has wrapped */
                        chunk->reader_pos = 0;
                        chunk->max_reader_pos = 0;
                        continue;
                    }

                    /* there is data to return */
                    break;
                }
                /* else, there is no data in this block */
                if (chunk->next) {
                    /* I don't think this ever gets called, since we
                     * should have moved to the next block when the reader
                     * returned its previous block */

                    /* free this block and move to the next */
                    chunk_pop();
                    continue;
                }
            }
            /* else no data is available */

            /* check the stopping condition */
            if (stopped || no_more_data) {
                return -1;
            }

            /* wait for data unless no_wait was specified */
            empty = true;
            if (no_wait) {
                return 1;
            }
            pthread_cond_wait(&cond, &mutex);
        }

        TRACE_FUNC;

        /* we get here when there is data to return */
        *reader_block = (conn_block_t*)&chunk->blocks[chunk->reader_pos];
        return 0;
    }

    void _return_read_block(
        const conn_block_t *reader_block)
    {
        conn_mem_chunk_t *chunk;

        assert(reader_block);

        chunk = reader_chunk;
        assert(reader_block==(conn_block_t*)&chunk->blocks[chunk->reader_pos]);

        /* move to next reader position */
        chunk->reader_pos += BLOCK_TOTAL_SIZE(reader_block->block_size);
        if (chunk->reader_pos == chunk->max_reader_pos) {
            chunk->reader_pos = 0;
            chunk->max_reader_pos = 0;
        }
        TRACE_MEM_CHUNK(chunk);

        /* check whether we need to move to the next chunk */
        if (chunk->reader_pos == chunk->writer_pos
            && chunk->next)
        {
            /* free this block and move to the next */
            chunk_pop();
        }
    }

    int _get_write_block(
        const conn_block_t    **writer_block,
        size_t                  size,
        bool                    no_wait)
    {
        conn_mem_chunk_t *chunk;
        size_t required_size;

        assert(writer_block);
        assert(size);

        *writer_block = NULL;

        if (block_max_size < size) {
            throw sk_error_conn_block_size_too_large_t();
        }

        /* account for overhead */
        required_size = BLOCK_TOTAL_SIZE(size);

        /* loop until there is space available */
        for (;;) {
            /* handle stopped condition */
            if (stopped) {
                return -1;
            }

            if (total_used + size > max_allocation) {
                /* we are full; wait for space unless no_wait was
                 * specified */
                full = true;
                if (no_wait) {
                    return 1;
                }
                pthread_cond_wait(&cond, &mutex);
                continue;
            }

            chunk = writer_chunk;
            if (NULL == chunk) {
                /* create the initial chunk */
                chunk = writer_chunk = reader_chunk = chunk_alloc();
            }

            if (chunk->writer_pos < chunk->reader_pos) {
                /* the writer has wrapped; look for space between the
                 * writer_pos and the reader_pos */
                if ((chunk->reader_pos - chunk->writer_pos)
                    >= (required_size + WRAP_GAP))
                {
                    /* space is available */
                    break;
                }
            } else {
                if ((chunk->capacity - chunk->writer_pos) >= required_size) {
                    /* space is available at end of the buffer */
                    break;
                }
                /* else, see if space is available at the front */
                if (chunk->reader_pos >= required_size + WRAP_GAP) {
                    /* space is available; wrap around */
                    chunk->max_reader_pos = chunk->writer_pos;
                    chunk->writer_pos = 0;
                    TRACE_MEM_CHUNK(chunk);
                    break;
                }
            }

            /* space is not available; allocate new block */
            chunk_push();
        }

        TRACE_FUNC;

        /* when we get here, space is available */
        *writer_block = (conn_block_t*)&chunk->blocks[chunk->writer_pos];
        ((conn_block_t*)&chunk->blocks[chunk->writer_pos])->block_size
            = required_size - sizeof(uint64_t);

        return  0;
    }

    void _return_write_block(
        const conn_block_t *writer_block,
        size_t              size)
    {
        conn_mem_chunk_t *chunk;

        assert(writer_block);
        assert(size > 0);

        chunk = writer_chunk;
        assert(writer_block==(conn_block_t*)&chunk->blocks[chunk->writer_pos]);

        ((conn_block_t*)&chunk->blocks[chunk->writer_pos])->block_size = size;
        chunk->writer_pos += BLOCK_TOTAL_SIZE(size);
        TRACE_MEM_CHUNK(chunk);
    }

    void _set_param(
        const Definition& defn,
        const std::vector<std::string>& values) throw (Error)
    {
        uint64_t u64;
        int rv;

        switch (*(param_id_t *)(defn.get_context())) {
          case CONN_MEM_CHUNK_SIZE:
            rv = skStringParseUint64(&u64, values[0].c_str(), 0, 0);
            if (rv) {
                goto PARSE_ERROR;
            }
            chunk_size = u64;
            break;

          case CONN_MEM_MAX_ALLOCATION:
            rv = skStringParseUint64(&u64, values[0].c_str(), 0, 0);
            if (rv) {
                goto PARSE_ERROR;
            }
            max_allocation = u64;
            break;
        }
        return;

      PARSE_ERROR:
        std::ostringstream os;
        os << "Invalid " << defn.get_name() << " '" << values[0] << "': "
           << skStringParseStrerror(rv);
        throw sk_error_conn_param_value_t(os.str());
    }


    void _init()
    {
        if (chunk_size) {
            if (chunk_size < CONN_MEM_MIN_CHUNK_SIZE) {
                std::ostringstream os;
                os << "Specified chunk size is below minimum of "
                   << CONN_MEM_MIN_CHUNK_SIZE;
                throw sk_error_conn_param_value_t(os.str());
            }
            if (0 == max_allocation) {
                /* use caller's chunk size, and the default multiple of it
                 * for the maximum allocation */
                max_allocation = CONN_MEM_STD_NUMBER_CHUNKS * chunk_size;
            } else if (chunk_size > max_allocation) {
                /* chunk size is larger than max allocation */
                throw sk_error_conn_param_value_t(
                    ("Specified chunk size greater than"
                     " specified max allocation"));
            }
            /* else both values specified and are valid */
        } else if (max_allocation) {
            if (max_allocation < CONN_MEM_MIN_CHUNK_SIZE) {
                std::ostringstream os;
                os << "Specified max allocation is below minimum of "
                   << CONN_MEM_MIN_CHUNK_SIZE;
                throw sk_error_conn_param_value_t(os.str());
            }

            /* determine the chunk size */
            if (max_allocation >= CONN_MEM_STD_CHUNK_SIZE) {
                /* use standard chunk size */
                chunk_size = CONN_MEM_STD_CHUNK_SIZE;
            } else {
                chunk_size = (max_allocation / CONN_MEM_STD_NUMBER_CHUNKS);
                if (chunk_size < CONN_MEM_MIN_CHUNK_SIZE) {
                    chunk_size = CONN_MEM_MIN_CHUNK_SIZE;
                }
            }
        } else {
            /* use the default sizes */
            chunk_size = CONN_MEM_STD_CHUNK_SIZE;
            max_allocation = CONN_MEM_STD_NUMBER_CHUNKS * chunk_size;
        }

        block_max_size = BLOCK_MAX_SIZE_FROM_CHUNK_SIZE(chunk_size);

        initialized = true;
    }

    /* linked list of chunks of memory */
    conn_mem_chunk_t   *writer_chunk;
    conn_mem_chunk_t   *reader_chunk;
    conn_mem_chunk_t   *spare_chunk;

    /* size of an individual chunk of memory; may be specified by the
     * caller */
    size_t              chunk_size;

    /* maximum block size a module may request; calculated as
     * roughly 1/3 of the chunk_size */
    size_t              block_max_size;

    enum param_id_t {
        CONN_MEM_CHUNK_SIZE,
        CONN_MEM_MAX_ALLOCATION
    };

    struct class_params_t {
        const char    *key;
        int            val;
        const char    *desciption;
    };

    static MemoryConnector::class_params_t class_params[] = {
        {"chunk_size",     CONN_MEM_CHUNK_SIZE,
         "Size of an individual chunk of memory"},
        {"max_allocation", CONN_MEM_MAX_ALLOCATION,
         ("Maximum amount of memory that may be allocated across all blocks"
          " on all chunks; does not include internal overhead")},
        {0, 0, 0}          /* sentinel */
    };

#if 0
    static const std::set<Parameterized::Definition>& create_param_defs()
    {
        size_t i;
        std::set<Parameterized::Definition> defs;

        for (i = 0; class_params[i].key != NULL; ++i) {
            assert(i == (size_t)class_params[i].val);
            Parameterized::Definition d(class_params[i].key,
                                        class_params[i].desciption,
                                        (void*)&class_params[i].val, 1, 1);
            assert(defs.find(d) = defs.end());
            defs.insert(d);
        }
        return defs;
    }
#endif

};

}

/*
** Local Variables:
** mode:c++
** indent-tabs-mode:nil
** c-basic-offset:4
** End:
*/