/*

 * Copyright (c) 2016-present, Yann Collet, Facebook, Inc.

 * All rights reserved.

 *

 * This source code is licensed under both the BSD-style license (found in the

 * LICENSE file in the root directory of this source tree) and the GPLv2 (found

 * in the COPYING file in the root directory of this source tree).

 * You may select, at your option, one of the above-listed licenses.

 */

/* ======   Compiler specifics   ====== */

#if defined(_MSC_VER)

#  pragma warning(disable : 4204)   /* disable: C4204: non-constant aggregate initializer */

#endif

/* ======   Constants   ====== */

#define ZSTDMT_OVERLAPLOG_DEFAULT 0

/* ======   Dependencies   ====== */

#include <string.h>      /* memcpy, memset */

#include <limits.h>      /* INT_MAX, UINT_MAX */

#include "pool.h"        /* threadpool */

#include "threading.h"   /* mutex */

#include "zstd_compress_internal.h"  /* MIN, ERROR, ZSTD_*, ZSTD_highbit32 */

#include "zstd_ldm.h"

#include "zstdmt_compress.h"

/* Guards code to support resizing the SeqPool.

 * We will want to resize the SeqPool to save memory in the future.

 * Until then, comment the code out since it is unused.

 */

#define ZSTD_RESIZE_SEQPOOL 0

/* ======   Debug   ====== */

#if defined(DEBUGLEVEL) && (DEBUGLEVEL>=2) \

    && !defined(_MSC_VER) \

    && !defined(__MINGW32__)

#  include <stdio.h>

#  include <unistd.h>

#  include <sys/times.h>

#  define DEBUG_PRINTHEX(l,p,n) {            \

    unsigned debug_u;                        \

    for (debug_u=0; debug_u<(n); debug_u++)  \

        RAWLOG(l, "%02X ", ((const unsigned char*)(p))[debug_u]); \

    RAWLOG(l, " \n");                        \

}

static unsigned long long GetCurrentClockTimeMicroseconds(void)

{

   static clock_t _ticksPerSecond = 0;

   if (_ticksPerSecond <= 0) _ticksPerSecond = sysconf(_SC_CLK_TCK);

   {   struct tms junk; clock_t newTicks = (clock_t) times(&junk);

       return ((((unsigned long long)newTicks)*(1000000))/_ticksPerSecond);

}  }

#define MUTEX_WAIT_TIME_DLEVEL 6

#define ZSTD_PTHREAD_MUTEX_LOCK(mutex) {          \

    if (DEBUGLEVEL >= MUTEX_WAIT_TIME_DLEVEL) {   \

        unsigned long long const beforeTime = GetCurrentClockTimeMicroseconds(); \

        ZSTD_pthread_mutex_lock(mutex);           \

        {   unsigned long long const afterTime = GetCurrentClockTimeMicroseconds(); \

            unsigned long long const elapsedTime = (afterTime-beforeTime); \

            if (elapsedTime > 1000) {  /* or whatever threshold you like; I'm using 1 millisecond here */ \

                DEBUGLOG(MUTEX_WAIT_TIME_DLEVEL, "Thread took %llu microseconds to acquire mutex %s \n", \

                   elapsedTime, #mutex);          \

        }   }                                     \

    } else {                                      \

        ZSTD_pthread_mutex_lock(mutex);           \

    }                                             \

}

#else

#  define ZSTD_PTHREAD_MUTEX_LOCK(m) ZSTD_pthread_mutex_lock(m)

#  define DEBUG_PRINTHEX(l,p,n) {}

#endif

/* =====   Buffer Pool   ===== */

/* a single Buffer Pool can be invoked from multiple threads in parallel */

typedef struct buffer_s {

    void* start;

    size_t capacity;

} buffer_t;

static const buffer_t g_nullBuffer = { NULL, 0 };

typedef struct ZSTDMT_bufferPool_s {

    ZSTD_pthread_mutex_t poolMutex;

    size_t bufferSize;

    unsigned totalBuffers;

    unsigned nbBuffers;

    ZSTD_customMem cMem;

    buffer_t bTable[1];   /* variable size */

} ZSTDMT_bufferPool;

static ZSTDMT_bufferPool* ZSTDMT_createBufferPool(unsigned nbWorkers, ZSTD_customMem cMem)

{

    unsigned const maxNbBuffers = 2*nbWorkers + 3;

    ZSTDMT_bufferPool* const bufPool = (ZSTDMT_bufferPool*)ZSTD_calloc(

        sizeof(ZSTDMT_bufferPool) + (maxNbBuffers-1) * sizeof(buffer_t), cMem);

    if (bufPool==NULL) return NULL;

    if (ZSTD_pthread_mutex_init(&bufPool->poolMutex, NULL)) {

        ZSTD_free(bufPool, cMem);

        return NULL;

    }

    bufPool->bufferSize = 64 KB;

    bufPool->totalBuffers = maxNbBuffers;

    bufPool->nbBuffers = 0;

    bufPool->cMem = cMem;

    return bufPool;

}

static void ZSTDMT_freeBufferPool(ZSTDMT_bufferPool* bufPool)

{

    unsigned u;

    DEBUGLOG(3, "ZSTDMT_freeBufferPool (address:%08X)", (U32)(size_t)bufPool);

    if (!bufPool) return;   /* compatibility with free on NULL */

    for (u=0; u<bufPool->totalBuffers; u++) {

        DEBUGLOG(4, "free buffer %2u (address:%08X)", u, (U32)(size_t)bufPool->bTable[u].start);

        ZSTD_free(bufPool->bTable[u].start, bufPool->cMem);

    }

    ZSTD_pthread_mutex_destroy(&bufPool->poolMutex);

    ZSTD_free(bufPool, bufPool->cMem);

}

/* only works at initialization, not during compression */

static size_t ZSTDMT_sizeof_bufferPool(ZSTDMT_bufferPool* bufPool)

{

    size_t const poolSize = sizeof(*bufPool)

                          + (bufPool->totalBuffers - 1) * sizeof(buffer_t);

    unsigned u;

    size_t totalBufferSize = 0;

    ZSTD_pthread_mutex_lock(&bufPool->poolMutex);

    for (u=0; u<bufPool->totalBuffers; u++)

        totalBufferSize += bufPool->bTable[u].capacity;

    ZSTD_pthread_mutex_unlock(&bufPool->poolMutex);

    return poolSize + totalBufferSize;

}

/* ZSTDMT_setBufferSize() :

 * all future buffers provided by this buffer pool will have _at least_ this size

 * note : it's better for all buffers to have same size,

 * as they become freely interchangeable, reducing malloc/free usages and memory fragmentation */

static void ZSTDMT_setBufferSize(ZSTDMT_bufferPool* const bufPool, size_t const bSize)

{

    ZSTD_pthread_mutex_lock(&bufPool->poolMutex);

    DEBUGLOG(4, "ZSTDMT_setBufferSize: bSize = %u", (U32)bSize);

    bufPool->bufferSize = bSize;

    ZSTD_pthread_mutex_unlock(&bufPool->poolMutex);

}

static ZSTDMT_bufferPool* ZSTDMT_expandBufferPool(ZSTDMT_bufferPool* srcBufPool, U32 nbWorkers)

{

    unsigned const maxNbBuffers = 2*nbWorkers + 3;

    if (srcBufPool==NULL) return NULL;

    if (srcBufPool->totalBuffers >= maxNbBuffers) /* good enough */

        return srcBufPool;

    /* need a larger buffer pool */

    {   ZSTD_customMem const cMem = srcBufPool->cMem;

        size_t const bSize = srcBufPool->bufferSize;   /* forward parameters */

        ZSTDMT_bufferPool* newBufPool;

        ZSTDMT_freeBufferPool(srcBufPool);

        newBufPool = ZSTDMT_createBufferPool(nbWorkers, cMem);

        if (newBufPool==NULL) return newBufPool;

        ZSTDMT_setBufferSize(newBufPool, bSize);

        return newBufPool;

    }

}

/** ZSTDMT_getBuffer() :

 *  assumption : bufPool must be valid

 * @return : a buffer, with start pointer and size

 *  note: allocation may fail, in this case, start==NULL and size==0 */

static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool)

{

    size_t const bSize = bufPool->bufferSize;

    DEBUGLOG(5, "ZSTDMT_getBuffer: bSize = %u", (U32)bufPool->bufferSize);

    ZSTD_pthread_mutex_lock(&bufPool->poolMutex);

    if (bufPool->nbBuffers) {   /* try to use an existing buffer */

        buffer_t const buf = bufPool->bTable[--(bufPool->nbBuffers)];

        size_t const availBufferSize = buf.capacity;

        bufPool->bTable[bufPool->nbBuffers] = g_nullBuffer;

        if ((availBufferSize >= bSize) & ((availBufferSize>>3) <= bSize)) {

            /* large enough, but not too much */

            DEBUGLOG(5, "ZSTDMT_getBuffer: provide buffer %u of size %u",

                        bufPool->nbBuffers, (U32)buf.capacity);

            ZSTD_pthread_mutex_unlock(&bufPool->poolMutex);

            return buf;

        }

        /* size conditions not respected : scratch this buffer, create new one */

        DEBUGLOG(5, "ZSTDMT_getBuffer: existing buffer does not meet size conditions => freeing");

        ZSTD_free(buf.start, bufPool->cMem);

    }

    ZSTD_pthread_mutex_unlock(&bufPool->poolMutex);

    /* create new buffer */

    DEBUGLOG(5, "ZSTDMT_getBuffer: create a new buffer");

    {   buffer_t buffer;

        void* const start = ZSTD_malloc(bSize, bufPool->cMem);

        buffer.start = start;   /* note : start can be NULL if malloc fails ! */

        buffer.capacity = (start==NULL) ? 0 : bSize;

        if (start==NULL) {

            DEBUGLOG(5, "ZSTDMT_getBuffer: buffer allocation failure !!");

        } else {

            DEBUGLOG(5, "ZSTDMT_getBuffer: created buffer of size %u", (U32)bSize);

        }

        return buffer;

    }

}

#if ZSTD_RESIZE_SEQPOOL

/** ZSTDMT_resizeBuffer() :

 * assumption : bufPool must be valid

 * @return : a buffer that is at least the buffer pool buffer size.

 *           If a reallocation happens, the data in the input buffer is copied.

 */

static buffer_t ZSTDMT_resizeBuffer(ZSTDMT_bufferPool* bufPool, buffer_t buffer)

{

    size_t const bSize = bufPool->bufferSize;

    if (buffer.capacity < bSize) {

        void* const start = ZSTD_malloc(bSize, bufPool->cMem);

        buffer_t newBuffer;

        newBuffer.start = start;

        newBuffer.capacity = start == NULL ? 0 : bSize;

        if (start != NULL) {

            assert(newBuffer.capacity >= buffer.capacity);

            memcpy(newBuffer.start, buffer.start, buffer.capacity);

            DEBUGLOG(5, "ZSTDMT_resizeBuffer: created buffer of size %u", (U32)bSize);

            return newBuffer;

        }

        DEBUGLOG(5, "ZSTDMT_resizeBuffer: buffer allocation failure !!");

    }

    return buffer;

}

#endif

/* store buffer for later re-use, up to pool capacity */

static void ZSTDMT_releaseBuffer(ZSTDMT_bufferPool* bufPool, buffer_t buf)

{

    DEBUGLOG(5, "ZSTDMT_releaseBuffer");

    if (buf.start == NULL) return;   /* compatible with release on NULL */

    ZSTD_pthread_mutex_lock(&bufPool->poolMutex);

    if (bufPool->nbBuffers < bufPool->totalBuffers) {

        bufPool->bTable[bufPool->nbBuffers++] = buf;  /* stored for later use */

        DEBUGLOG(5, "ZSTDMT_releaseBuffer: stored buffer of size %u in slot %u",

                    (U32)buf.capacity, (U32)(bufPool->nbBuffers-1));

        ZSTD_pthread_mutex_unlock(&bufPool->poolMutex);

        return;

    }

    ZSTD_pthread_mutex_unlock(&bufPool->poolMutex);

    /* Reached bufferPool capacity (should not happen) */

    DEBUGLOG(5, "ZSTDMT_releaseBuffer: pool capacity reached => freeing ");

    ZSTD_free(buf.start, bufPool->cMem);

}

/* =====   Seq Pool Wrapper   ====== */

static rawSeqStore_t kNullRawSeqStore = {NULL, 0, 0, 0};

typedef ZSTDMT_bufferPool ZSTDMT_seqPool;

static size_t ZSTDMT_sizeof_seqPool(ZSTDMT_seqPool* seqPool)

{

    return ZSTDMT_sizeof_bufferPool(seqPool);

}

static rawSeqStore_t bufferToSeq(buffer_t buffer)

{

    rawSeqStore_t seq = {NULL, 0, 0, 0};

    seq.seq = (rawSeq*)buffer.start;

    seq.capacity = buffer.capacity / sizeof(rawSeq);

    return seq;

}

static buffer_t seqToBuffer(rawSeqStore_t seq)

{

    buffer_t buffer;

    buffer.start = seq.seq;

    buffer.capacity = seq.capacity * sizeof(rawSeq);

    return buffer;

}

static rawSeqStore_t ZSTDMT_getSeq(ZSTDMT_seqPool* seqPool)

{

    if (seqPool->bufferSize == 0) {

        return kNullRawSeqStore;

    }

    return bufferToSeq(ZSTDMT_getBuffer(seqPool));

}

#if ZSTD_RESIZE_SEQPOOL

static rawSeqStore_t ZSTDMT_resizeSeq(ZSTDMT_seqPool* seqPool, rawSeqStore_t seq)

{

  return bufferToSeq(ZSTDMT_resizeBuffer(seqPool, seqToBuffer(seq)));

}

#endif

static void ZSTDMT_releaseSeq(ZSTDMT_seqPool* seqPool, rawSeqStore_t seq)

{

  ZSTDMT_releaseBuffer(seqPool, seqToBuffer(seq));

}

static void ZSTDMT_setNbSeq(ZSTDMT_seqPool* const seqPool, size_t const nbSeq)

{

  ZSTDMT_setBufferSize(seqPool, nbSeq * sizeof(rawSeq));

}

static ZSTDMT_seqPool* ZSTDMT_createSeqPool(unsigned nbWorkers, ZSTD_customMem cMem)

{

    ZSTDMT_seqPool* const seqPool = ZSTDMT_createBufferPool(nbWorkers, cMem);

    if (seqPool == NULL) return NULL;

    ZSTDMT_setNbSeq(seqPool, 0);

    return seqPool;

}

static void ZSTDMT_freeSeqPool(ZSTDMT_seqPool* seqPool)

{

    ZSTDMT_freeBufferPool(seqPool);

}

static ZSTDMT_seqPool* ZSTDMT_expandSeqPool(ZSTDMT_seqPool* pool, U32 nbWorkers)

{

    return ZSTDMT_expandBufferPool(pool, nbWorkers);

}

/* =====   CCtx Pool   ===== */

/* a single CCtx Pool can be invoked from multiple threads in parallel */

typedef struct {

    ZSTD_pthread_mutex_t poolMutex;

    int totalCCtx;

    int availCCtx;

    ZSTD_customMem cMem;

    ZSTD_CCtx* cctx[1];   /* variable size */

} ZSTDMT_CCtxPool;

/* note : all CCtx borrowed from the pool should be released back to the pool _before_ freeing the pool */

static void ZSTDMT_freeCCtxPool(ZSTDMT_CCtxPool* pool)

{

    int cid;

    for (cid=0; cid<pool->totalCCtx; cid++)

        ZSTD_freeCCtx(pool->cctx[cid]);  /* note : compatible with free on NULL */

    ZSTD_pthread_mutex_destroy(&pool->poolMutex);

    ZSTD_free(pool, pool->cMem);

}

/* ZSTDMT_createCCtxPool() :

 * implies nbWorkers >= 1 , checked by caller ZSTDMT_createCCtx() */

static ZSTDMT_CCtxPool* ZSTDMT_createCCtxPool(int nbWorkers,

                                              ZSTD_customMem cMem)

{

    ZSTDMT_CCtxPool* const cctxPool = (ZSTDMT_CCtxPool*) ZSTD_calloc(

        sizeof(ZSTDMT_CCtxPool) + (nbWorkers-1)*sizeof(ZSTD_CCtx*), cMem);

    assert(nbWorkers > 0);

    if (!cctxPool) return NULL;

    if (ZSTD_pthread_mutex_init(&cctxPool->poolMutex, NULL)) {

        ZSTD_free(cctxPool, cMem);

        return NULL;

    }

    cctxPool->cMem = cMem;

    cctxPool->totalCCtx = nbWorkers;

    cctxPool->availCCtx = 1;   /* at least one cctx for single-thread mode */

    cctxPool->cctx[0] = ZSTD_createCCtx_advanced(cMem);

    if (!cctxPool->cctx[0]) { ZSTDMT_freeCCtxPool(cctxPool); return NULL; }

    DEBUGLOG(3, "cctxPool created, with %u workers", nbWorkers);

    return cctxPool;

}

static ZSTDMT_CCtxPool* ZSTDMT_expandCCtxPool(ZSTDMT_CCtxPool* srcPool,

                                              int nbWorkers)

{

    if (srcPool==NULL) return NULL;

    if (nbWorkers <= srcPool->totalCCtx) return srcPool;   /* good enough */

    /* need a larger cctx pool */

    {   ZSTD_customMem const cMem = srcPool->cMem;

        ZSTDMT_freeCCtxPool(srcPool);

        return ZSTDMT_createCCtxPool(nbWorkers, cMem);

    }

}

/* only works during initialization phase, not during compression */

static size_t ZSTDMT_sizeof_CCtxPool(ZSTDMT_CCtxPool* cctxPool)

{

    ZSTD_pthread_mutex_lock(&cctxPool->poolMutex);

    {   unsigned const nbWorkers = cctxPool->totalCCtx;

        size_t const poolSize = sizeof(*cctxPool)

                                + (nbWorkers-1) * sizeof(ZSTD_CCtx*);

        unsigned u;

        size_t totalCCtxSize = 0;

        for (u=0; u<nbWorkers; u++) {

            totalCCtxSize += ZSTD_sizeof_CCtx(cctxPool->cctx[u]);

        }

        ZSTD_pthread_mutex_unlock(&cctxPool->poolMutex);

        assert(nbWorkers > 0);

        return poolSize + totalCCtxSize;

    }

}

static ZSTD_CCtx* ZSTDMT_getCCtx(ZSTDMT_CCtxPool* cctxPool)

{

    DEBUGLOG(5, "ZSTDMT_getCCtx");

    ZSTD_pthread_mutex_lock(&cctxPool->poolMutex);

    if (cctxPool->availCCtx) {

        cctxPool->availCCtx--;

        {   ZSTD_CCtx* const cctx = cctxPool->cctx[cctxPool->availCCtx];

            ZSTD_pthread_mutex_unlock(&cctxPool->poolMutex);

            return cctx;

    }   }

    ZSTD_pthread_mutex_unlock(&cctxPool->poolMutex);

    DEBUGLOG(5, "create one more CCtx");

    return ZSTD_createCCtx_advanced(cctxPool->cMem);   /* note : can be NULL, when creation fails ! */

}

static void ZSTDMT_releaseCCtx(ZSTDMT_CCtxPool* pool, ZSTD_CCtx* cctx)

{

    if (cctx==NULL) return;   /* compatibility with release on NULL */

    ZSTD_pthread_mutex_lock(&pool->poolMutex);

    if (pool->availCCtx < pool->totalCCtx)

        pool->cctx[pool->availCCtx++] = cctx;

    else {

        /* pool overflow : should not happen, since totalCCtx==nbWorkers */

        DEBUGLOG(4, "CCtx pool overflow : free cctx");

        ZSTD_freeCCtx(cctx);

    }

    ZSTD_pthread_mutex_unlock(&pool->poolMutex);

}

/* ====   Serial State   ==== */

typedef struct {

    void const* start;

    size_t size;

} range_t;

typedef struct {

    /* All variables in the struct are protected by mutex. */

    ZSTD_pthread_mutex_t mutex;

    ZSTD_pthread_cond_t cond;

    ZSTD_CCtx_params params;

    ldmState_t ldmState;

    XXH64_state_t xxhState;

    unsigned nextJobID;

    /* Protects ldmWindow.

     * Must be acquired after the main mutex when acquiring both.

     */

    ZSTD_pthread_mutex_t ldmWindowMutex;

    ZSTD_window_t ldmWindow;  /* A thread-safe copy of ldmState.window */

} serialState_t;

static int ZSTDMT_serialState_reset(serialState_t* serialState, ZSTDMT_seqPool* seqPool, ZSTD_CCtx_params params, size_t jobSize)

{

    /* Adjust parameters */

    if (params.ldmParams.enableLdm) {

        DEBUGLOG(4, "LDM window size = %u KB", (1U << params.cParams.windowLog) >> 10);

        ZSTD_ldm_adjustParameters(&params.ldmParams, &params.cParams);

        assert(params.ldmParams.hashLog >= params.ldmParams.bucketSizeLog);

        assert(params.ldmParams.hashRateLog < 32);

        serialState->ldmState.hashPower =

                ZSTD_rollingHash_primePower(params.ldmParams.minMatchLength);

    } else {

        memset(&params.ldmParams, 0, sizeof(params.ldmParams));

    }

    serialState->nextJobID = 0;

    if (params.fParams.checksumFlag)

        XXH64_reset(&serialState->xxhState, 0);

    if (params.ldmParams.enableLdm) {

        ZSTD_customMem cMem = params.customMem;

        unsigned const hashLog = params.ldmParams.hashLog;

        size_t const hashSize = ((size_t)1 << hashLog) * sizeof(ldmEntry_t);

        unsigned const bucketLog =

            params.ldmParams.hashLog - params.ldmParams.bucketSizeLog;

        size_t const bucketSize = (size_t)1 << bucketLog;

        unsigned const prevBucketLog =

            serialState->params.ldmParams.hashLog -

            serialState->params.ldmParams.bucketSizeLog;

        /* Size the seq pool tables */

        ZSTDMT_setNbSeq(seqPool, ZSTD_ldm_getMaxNbSeq(params.ldmParams, jobSize));

        /* Reset the window */

        ZSTD_window_clear(&serialState->ldmState.window);

        serialState->ldmWindow = serialState->ldmState.window;