/*
    Copyright (c) 2014-2015 Intel Corporation.  All Rights Reserved.

    Redistribution and use in source and binary forms, with or without
    modification, are permitted provided that the following conditions
    are met:

      * Redistributions of source code must retain the above copyright
        notice, this list of conditions and the following disclaimer.
      * Redistributions in binary form must reproduce the above copyright
        notice, this list of conditions and the following disclaimer in the
        documentation and/or other materials provided with the distribution.
      * Neither the name of Intel Corporation nor the names of its
        contributors may be used to endorse or promote products derived
        from this software without specific prior written permission.

    THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
    "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
    LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
    A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
    HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
    SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
    LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
    DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
    THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
    (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
    OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/


#include "offload_engine.h"
#include <signal.h>
#include <errno.h>

#include <algorithm>
#include <vector>

#include "offload_host.h"
#include "offload_table.h"
#include "offload_iterator.h"

// Static members of Stream class must be described somewhere.
// This members describe the list of all streams defined in programm
// via call to _Offload_stream_create.
uint64_t  Stream::m_streams_count = 0;
StreamMap Stream::all_streams;
mutex_t   Stream::m_stream_lock;

const char* Engine::m_func_names[Engine::c_funcs_total] =
{
    "server_compute",
#ifdef MYO_SUPPORT
    "server_myoinit",
    "server_myofini",
#endif // MYO_SUPPORT
    "server_init",
    "server_var_table_size",
    "server_var_table_copy",
    "server_set_stream_affinity"
};

// Symbolic representation of system signals. Fix for CQ233593
const char* Engine::c_signal_names[Engine::c_signal_max] =
{
    "Unknown SIGNAL",
    "SIGHUP",    /*  1, Hangup (POSIX).  */
    "SIGINT",    /*  2, Interrupt (ANSI).  */
    "SIGQUIT",   /*  3, Quit (POSIX).  */
    "SIGILL",    /*  4, Illegal instruction (ANSI).  */
    "SIGTRAP",   /*  5, Trace trap (POSIX).  */
    "SIGABRT",   /*  6, Abort (ANSI).  */
    "SIGBUS",    /*  7, BUS error (4.2 BSD).  */
    "SIGFPE",    /*  8, Floating-point exception (ANSI).  */
    "SIGKILL",   /*  9, Kill, unblockable (POSIX).  */
    "SIGUSR1",   /* 10, User-defined signal 1 (POSIX).  */
    "SIGSEGV",   /* 11, Segmentation violation (ANSI).  */
    "SIGUSR2",   /* 12, User-defined signal 2 (POSIX).  */
    "SIGPIPE",   /* 13, Broken pipe (POSIX).  */
    "SIGALRM",   /* 14, Alarm clock (POSIX).  */
    "SIGTERM",   /* 15, Termination (ANSI).  */
    "SIGSTKFLT", /* 16, Stack fault.  */
    "SIGCHLD",   /* 17, Child status has changed (POSIX).  */
    "SIGCONT",   /* 18, Continue (POSIX).  */
    "SIGSTOP",   /* 19, Stop, unblockable (POSIX).  */
    "SIGTSTP",   /* 20, Keyboard stop (POSIX).  */
    "SIGTTIN",   /* 21, Background read from tty (POSIX).  */
    "SIGTTOU",   /* 22, Background write to tty (POSIX).  */
    "SIGURG",    /* 23, Urgent condition on socket (4.2 BSD).  */
    "SIGXCPU",   /* 24, CPU limit exceeded (4.2 BSD).  */
    "SIGXFSZ",   /* 25, File size limit exceeded (4.2 BSD).  */
    "SIGVTALRM", /* 26, Virtual alarm clock (4.2 BSD).  */
    "SIGPROF",   /* 27, Profiling alarm clock (4.2 BSD).  */
    "SIGWINCH",  /* 28, Window size change (4.3 BSD, Sun).  */
    "SIGIO",     /* 29, I/O now possible (4.2 BSD).  */
    "SIGPWR",    /* 30, Power failure restart (System V).  */
    "SIGSYS"     /* 31, Bad system call.  */
};

void Engine::init(void)
{
    if (!m_ready) {
        mutex_locker_t locker(m_lock);

        if (!m_ready) {
            // start process if not done yet
            if (m_process == 0) {
                init_process();
            }

            // load penging images
            load_libraries();

            // and (re)build pointer table
            init_ptr_data();

            // it is ready now
            m_ready = true;
        }
    }
}

void Engine::init_process(void)
{
    COIENGINE engine;
    COIRESULT res;
    const char **environ;
    char buf[4096];  // For exe path name

    // create environment for the target process
    environ = (const char**) mic_env_vars.create_environ_for_card(m_index);
    if (environ != 0) {
        for (const char **p = environ; *p != 0; p++) {
            OFFLOAD_DEBUG_TRACE(3, "Env Var for card %d: %s\n", m_index, *p);
        }
    }

    // Create execution context in the specified device
    OFFLOAD_DEBUG_TRACE(2, "Getting device %d (engine %d) handle\n", m_index,
                        m_physical_index);
    res = COI::EngineGetHandle(COI_ISA_MIC, m_physical_index, &engine);
    check_result(res, c_get_engine_handle, m_index, res);

    // Get engine info on threads and cores.
    // The values of core number and thread number will be used later at stream
    // creation by call to _Offload_stream_create(device,number_of_cpus).

    COI_ENGINE_INFO engine_info;

    res = COI::EngineGetInfo(engine, sizeof(COI_ENGINE_INFO), &engine_info);
    check_result(res, c_get_engine_info, m_index, res);

    // m_cpus bitset has 1 for available thread. At the begining all threads
    // are available and m_cpus(i) is set to
    // 1 for i = [0...engine_info.NumThreads].
    m_cpus.reset();
    for (int i = 0; i < engine_info.NumThreads; i++) {
         m_cpus.set(i);
    }

    // The following values will be used at pipeline creation for streams
    m_num_cores = engine_info.NumCores;
    m_num_threads = engine_info.NumThreads;

    // Check if OFFLOAD_DMA_CHANNEL_COUNT is set to 2
    // Only the value 2 is supported in 16.0
    if (mic_dma_channel_count == 2) {
        if (COI::ProcessConfigureDMA) {
            // Set DMA channels using COI API
            COI::ProcessConfigureDMA(2, COI::DMA_MODE_READ_WRITE);
        }
        else {
            // Set environment variable COI_DMA_CHANNEL_COUNT
            // use putenv instead of setenv as Windows has no setenv.
            // Note: putenv requires its argument can't be freed or modified.
            // So no free after call to putenv or elsewhere.
            char * env_var = strdup("COI_DMA_CHANNEL_COUNT=2");
	    if (env_var == NULL)
	      LIBOFFLOAD_ERROR(c_malloc);
            putenv(env_var);  
        }
    }

    // Target executable is not available then use compiler provided offload_main
    if (__target_exe == 0) {
       if (mic_device_main == 0)
          LIBOFFLOAD_ERROR(c_report_no_host_exe);

       OFFLOAD_DEBUG_TRACE(2,
           "Loading target executable %s\n",mic_device_main);

       res = COI::ProcessCreateFromFile(
           engine,                 // in_Engine
           mic_device_main,        // in_pBinaryName
           0,                      // in_Argc
           0,                      // in_ppArgv
           environ == 0,           // in_DupEnv
           environ,                // in_ppAdditionalEnv
           mic_proxy_io,           // in_ProxyActive
           mic_proxy_fs_root,      // in_ProxyfsRoot
           mic_buffer_size,        // in_BufferSpace
           mic_library_path,       // in_LibrarySearchPath
           &m_process              // out_pProcess
       );
    }
    else {
    // Target executable should be available by the time when we
    // attempt to initialize the device

       //  Need the full path of the FAT exe for VTUNE
       {
#ifndef TARGET_WINNT
          ssize_t len = readlink("/proc/self/exe", buf,1000);
#else
          int len = GetModuleFileName(NULL, buf,1000);
#endif // TARGET_WINNT
          if  (len == -1) {
             LIBOFFLOAD_ERROR(c_report_no_host_exe);
             exit(1);
          }
          else if (len > 999) {
             LIBOFFLOAD_ERROR(c_report_path_buff_overflow);
             exit(1);
          }
          buf[len] = '\0';
       }

       OFFLOAD_DEBUG_TRACE(2,
           "Loading target executable \"%s\" from %p, size %lld, host file %s\n",
           __target_exe->name, __target_exe->data, __target_exe->size,
           buf);

       res = COI::ProcessCreateFromMemory(
           engine,                 // in_Engine
           __target_exe->name,     // in_pBinaryName
           __target_exe->data,     // in_pBinaryBuffer
           __target_exe->size,     // in_BinaryBufferLength,
           0,                      // in_Argc
           0,                      // in_ppArgv
           environ == 0,           // in_DupEnv
           environ,                // in_ppAdditionalEnv
           mic_proxy_io,           // in_ProxyActive
           mic_proxy_fs_root,      // in_ProxyfsRoot
           mic_buffer_size,        // in_BufferSpace
           mic_library_path,       // in_LibrarySearchPath
           buf,                    // in_FileOfOrigin
           -1,                     // in_FileOfOriginOffset use -1 to indicate to
                                   // COI that is is a FAT binary
           &m_process              // out_pProcess
       );
    }
    check_result(res, c_process_create, m_index, res);

    if ((mic_4k_buffer_size != 0) || (mic_2m_buffer_size !=0)) {
       // available only in MPSS 4.2 and greater
       if (COI::ProcessSetCacheSize != 0 ) { 
          int flags;
          //  Need compiler to use MPSS 3.2 or greater to get these
          // definition so currently hardcoding it
          //  COI_CACHE_ACTION_GROW_NOW && COI_CACHE_MODE_ONDEMAND_SYNC;
          flags = 0x00020002; 
          res = COI::ProcessSetCacheSize(
               m_process,             // in_Process
               mic_2m_buffer_size,    // in_HugePagePoolSize
               flags,                 // inHugeFlags
               mic_4k_buffer_size,    // in_SmallPagePoolSize
               flags,                 // inSmallFlags
               0,                     // in_NumDependencies
               0,                     // in_pDependencies
               0                      // out_PCompletion
          );
          OFFLOAD_DEBUG_TRACE(2,
              "Reserve target buffers 4K pages = %d  2M pages = %d\n",
                  mic_4k_buffer_size, mic_2m_buffer_size);
           check_result(res, c_process_set_cache_size, m_index, res);
       }
       else {
             OFFLOAD_DEBUG_TRACE(2,
                 "Reserve target buffers not supported in current MPSS\n");
       }
    }

    // get function handles
    res = COI::ProcessGetFunctionHandles(m_process, c_funcs_total,
                                         m_func_names, m_funcs);
    check_result(res, c_process_get_func_handles, m_index, res);

    // initialize device side
    pid_t pid = init_device();

    // For IDB
    if (__dbg_is_attached) {
        // TODO: we have in-memory executable now.
        // Check with IDB team what should we provide them now?
        if (strlen(__target_exe->name) < MAX_TARGET_NAME) {
            strcpy(__dbg_target_exe_name, __target_exe->name);
        }
        __dbg_target_so_pid = pid;
        __dbg_target_id = m_physical_index;
        __dbg_target_so_loaded();
    }
}

void Engine::fini_process(bool verbose)
{
    if (m_process != 0) {
        uint32_t sig;
        int8_t ret;

        // destroy target process
        OFFLOAD_DEBUG_TRACE(2, "Destroying process on the device %d\n",
                            m_index);

        COIRESULT res = COI::ProcessDestroy(m_process, -1, 0, &ret, &sig);
        m_process = 0;

        if (res == COI_SUCCESS) {
            OFFLOAD_DEBUG_TRACE(3, "Device process: signal %d, exit code %d\n",
                                sig, ret);
            if (verbose) {
                if (sig != 0) {
                    LIBOFFLOAD_ERROR(
                        c_mic_process_exit_sig, m_index, sig,
                        c_signal_names[sig >= c_signal_max ? 0 : sig]);
                }
                else {
                    LIBOFFLOAD_ERROR(c_mic_process_exit_ret, m_index, ret);
                }
            }

            // for idb
            if (__dbg_is_attached) {
                __dbg_target_so_unloaded();
            }
        }
        else {
            if (verbose) {
                LIBOFFLOAD_ERROR(c_mic_process_exit, m_index);
            }
        }
    }
}

void Engine::load_libraries()
{
    // load libraries collected so far
    for (TargetImageList::iterator it = m_images.begin();
         it != m_images.end(); it++) {
        OFFLOAD_DEBUG_TRACE(2,
            "Loading library \"%s\" from %p, size %llu, host file %s\n",
                                    it->name, it->data, it->size, it->origin);

        // load library to the device
        COILIBRARY lib;
        COIRESULT res;
        res = COI::ProcessLoadLibraryFromMemory(m_process,
                                                it->data,
                                                it->size,
                                                it->name,
                                                mic_library_path,
                                                it->origin,
                                                (it->origin) ? -1 : 0,
                                                COI_LOADLIBRARY_V1_FLAGS,
                                                &lib);
        m_dyn_libs.push_front(DynLib(it->name, it->data, lib));

        if (res != COI_SUCCESS && res != COI_ALREADY_EXISTS) {
            check_result(res, c_load_library, m_index, res);
        }
    }
    m_images.clear();
}

void Engine::unload_library(const void *data, const char *name)
{
    if (m_process == 0) {
       return;
    }
    for (DynLibList::iterator it = m_dyn_libs.begin();
         it != m_dyn_libs.end(); it++) {
         if (it->data == data) {
            COIRESULT res;
            OFFLOAD_DEBUG_TRACE(2,
               "Unloading library \"%s\"\n",name);
            res = COI::ProcessUnloadLibrary(m_process,it->lib);
            m_dyn_libs.erase(it);
            if (res != COI_SUCCESS) {
                check_result(res, c_unload_library, m_index, res);
            }
            return;
         }
    }
}

static bool target_entry_cmp(
    const VarList::BufEntry &l,
    const VarList::BufEntry &r
)
{
    const char *l_name = reinterpret_cast<const char*>(l.name);
    const char *r_name = reinterpret_cast<const char*>(r.name);
    return strcmp(l_name, r_name) < 0;
}

static bool host_entry_cmp(
    const VarTable::Entry *l,
    const VarTable::Entry *r
)
{
    return strcmp(l->name, r->name) < 0;
}

void Engine::init_ptr_data(void)
{
    COIRESULT res;
    COIEVENT event;

    // Prepare table of host entries
    std::vector<const VarTable::Entry*> host_table(
                                         Iterator(__offload_vars.get_head()),
                                         Iterator());

    // no need to do anything further is host table is empty
    if (host_table.size() <= 0) {
        return;
    }

    // Get var table entries from the target.
    // First we need to get size for the buffer to copy data
    struct {
        int64_t nelems;
        int64_t length;
    } params;

    res = COI::PipelineRunFunction(get_pipeline(),
                                   m_funcs[c_func_var_table_size],
                                   0, 0, 0,
                                   0, 0,
                                   0, 0,
                                   &params, sizeof(params),
                                   &event);
    check_result(res, c_pipeline_run_func, m_index, res);

    res = COI::EventWait(1, &event, -1, 1, 0, 0);
    check_result(res, c_event_wait, res);

    if (params.length == 0) {
        return;
    }

    // create buffer for target entries and copy data to host
    COIBUFFER buffer;
    res = COI::BufferCreate(params.length, COI_BUFFER_NORMAL, 0, 0, 1,
                            &m_process, &buffer);
    check_result(res, c_buf_create, m_index, res);

    COI_ACCESS_FLAGS flags = COI_SINK_WRITE;
    res = COI::PipelineRunFunction(get_pipeline(),
                                   m_funcs[c_func_var_table_copy],
                                   1, &buffer, &flags,
                                   0, 0,
                                   &params.nelems, sizeof(params.nelems),
                                   0, 0,
                                   &event);
    check_result(res, c_pipeline_run_func, m_index, res);

    res = COI::EventWait(1, &event, -1, 1, 0, 0);
    check_result(res, c_event_wait, res);

    // patch names in target data
    VarList::BufEntry *target_table;
    COIMAPINSTANCE map_inst;
    res = COI::BufferMap(buffer, 0, params.length, COI_MAP_READ_ONLY, 0, 0,
                         0, &map_inst,
                         reinterpret_cast<void**>(&target_table));
    check_result(res, c_buf_map, res);

    VarList::table_patch_names(target_table, params.nelems);

    // and sort entries
    std::sort(target_table, target_table + params.nelems, target_entry_cmp);
    std::sort(host_table.begin(), host_table.end(), host_entry_cmp);

    // merge host and target entries and enter matching vars map
    std::vector<const VarTable::Entry*>::const_iterator hi =
        host_table.begin();
    std::vector<const VarTable::Entry*>::const_iterator he =
        host_table.end();
    const VarList::BufEntry *ti = target_table;
    const VarList::BufEntry *te = target_table + params.nelems;

    while (hi != he && ti != te) {
        int res = strcmp((*hi)->name, reinterpret_cast<const char*>(ti->name));
        if (res == 0) {
            bool is_new;
            // add matching entry to var map
            PtrData *ptr = insert_ptr_data((*hi)->addr, (*hi)->size, is_new);

            // store address for new entries
            if (is_new) {
                ptr->mic_addr = ti->addr;
                ptr->is_static = true;
            }
            ptr->alloc_ptr_data_lock.unlock();
            hi++;
            ti++;
        }
        else if (res < 0) {
            hi++;
        }
        else {
            ti++;
        }
    }

    // cleanup
    res = COI::BufferUnmap(map_inst, 0, 0, 0);
    check_result(res, c_buf_unmap, res);

    res = COI::BufferDestroy(buffer);
    check_result(res, c_buf_destroy, res);
}

COIRESULT Engine::compute(
    _Offload_stream stream,
    const std::list<COIBUFFER> &buffers,
    const void*         data,
    uint16_t            data_size,
    void*               ret,
    uint16_t            ret_size,
    uint32_t            num_deps,
    const COIEVENT*     deps,
    COIEVENT*           event
) /* const */
{
    COIBUFFER *bufs;
    COI_ACCESS_FLAGS *flags;
    COIRESULT res;

    // convert buffers list to array
    int num_bufs = buffers.size();
    if (num_bufs > 0) {
        bufs = (COIBUFFER*) alloca(num_bufs * sizeof(COIBUFFER));
        flags = (COI_ACCESS_FLAGS*) alloca(num_bufs *
                                           sizeof(COI_ACCESS_FLAGS));

        int i = 0;
        for (std::list<COIBUFFER>::const_iterator it = buffers.begin();
             it != buffers.end(); it++) {
            bufs[i] = *it;

            // TODO: this should be fixed
            flags[i++] = COI_SINK_WRITE;
        }
    }
    else {
        bufs = 0;
        flags = 0;
    }
    COIPIPELINE pipeline = (stream == no_stream) ?
                           get_pipeline() :
                           get_pipeline(stream);
    // start computation
    res = COI::PipelineRunFunction(pipeline,
                                   m_funcs[c_func_compute],
                                   num_bufs, bufs, flags,
                                   num_deps, deps,
                                   data, data_size,
                                   ret, ret_size,
                                   event);
    return res;
}

pid_t Engine::init_device(void)
{
    struct init_data {
        int  device_index;
        int  devices_total;
        int  console_level;
        int  offload_report_level;
    } data;
    COIRESULT res;
    COIEVENT event;
    pid_t pid;

    OFFLOAD_DEBUG_TRACE_1(2, 0, c_offload_init,
                          "Initializing device with logical index %d "
                          "and physical index %d\n",
                           m_index, m_physical_index);

    // setup misc data
    data.device_index = m_index;
    data.devices_total = mic_engines_total;
    data.console_level = console_enabled;
    data.offload_report_level = offload_report_level;

    res = COI::PipelineRunFunction(get_pipeline(),
                                   m_funcs[c_func_init],
                                   0, 0, 0, 0, 0,
                                   &data, sizeof(data),
                                   &pid, sizeof(pid),
                                   &event);
    check_result(res, c_pipeline_run_func, m_index, res);

    res = COI::EventWait(1, &event, -1, 1, 0, 0);
    check_result(res, c_event_wait, res);

    OFFLOAD_DEBUG_TRACE(2, "Device process pid is %d\n", pid);

    return pid;
}

// data associated with each thread
struct Thread {
    Thread(long* addr_coipipe_counter) {
        m_addr_coipipe_counter = addr_coipipe_counter;
        memset(m_pipelines, 0, sizeof(m_pipelines));
    }

    ~Thread() {
#ifndef TARGET_WINNT
        __sync_sub_and_fetch(m_addr_coipipe_counter, 1);
#else // TARGET_WINNT
        _InterlockedDecrement(m_addr_coipipe_counter);
#endif // TARGET_WINNT
        for (int i = 0; i < mic_engines_total; i++) {
            if (m_pipelines[i] != 0) {
                COI::PipelineDestroy(m_pipelines[i]);
            }
        }
    }

    COIPIPELINE get_pipeline(int index) const {
        return m_pipelines[index];
    }

    void set_pipeline(int index, COIPIPELINE pipeline) {
        m_pipelines[index] = pipeline;
    }

    AutoSet& get_auto_vars() {
        return m_auto_vars;
    }

private:
    long*       m_addr_coipipe_counter;
    AutoSet     m_auto_vars;
    COIPIPELINE m_pipelines[MIC_ENGINES_MAX];
};

COIPIPELINE Engine::get_pipeline(void)
{
    Thread* thread = (Thread*) thread_getspecific(mic_thread_key);
    if (thread == 0) {
        thread = new Thread(&m_proc_number);
        thread_setspecific(mic_thread_key, thread);
    }

    COIPIPELINE pipeline = thread->get_pipeline(m_index);
    if (pipeline == 0) {
        COIRESULT res;
        int proc_num;

#ifndef TARGET_WINNT
        proc_num = __sync_fetch_and_add(&m_proc_number, 1);
#else // TARGET_WINNT
        proc_num = _InterlockedIncrement(&m_proc_number);
#endif // TARGET_WINNT

        if (proc_num > COI_PIPELINE_MAX_PIPELINES) {
            LIBOFFLOAD_ERROR(c_coipipe_max_number, COI_PIPELINE_MAX_PIPELINES);
            LIBOFFLOAD_ABORT;
        }
        // create pipeline for this thread
        res = COI::PipelineCreate(m_process, 0, mic_stack_size, &pipeline);
        check_result(res, c_pipeline_create, m_index, res);
        thread->set_pipeline(m_index, pipeline);
    }
    return pipeline;
}

Stream* Stream::find_stream(uint64_t handle, bool remove)
{
    Stream *stream = 0;

    m_stream_lock.lock();
    {
        StreamMap::iterator it = all_streams.find(handle);
        if (it != all_streams.end()) {
            stream = it->second;
            if (remove) {
                all_streams.erase(it);
            }
        }
    }
    m_stream_lock.unlock();
    return stream;
}

COIPIPELINE Engine::get_pipeline(_Offload_stream handle)
{
    Stream * stream = Stream::find_stream(handle, false);

    if (!stream) {
        LIBOFFLOAD_ERROR(c_offload_no_stream, m_index);
        LIBOFFLOAD_ABORT;
    }

    COIPIPELINE pipeline = stream->get_pipeline();

    if (pipeline == 0) {
        COIRESULT     res;
        int           proc_num;
        COI_CPU_MASK  in_Mask ;

#ifndef TARGET_WINNT
        proc_num = __sync_fetch_and_add(&m_proc_number, 1);
#else // TARGET_WINNT
        proc_num = _InterlockedIncrement(&m_proc_number);
#endif // TARGET_WINNT

        if (proc_num > COI_PIPELINE_MAX_PIPELINES) {
            LIBOFFLOAD_ERROR(c_coipipe_max_number, COI_PIPELINE_MAX_PIPELINES);
            LIBOFFLOAD_ABORT;
        }

        m_stream_lock.lock();

        // start process if not done yet
        if (m_process == 0) {
            init_process();
        }

        // create CPUmask
        res = COI::PipelineClearCPUMask(in_Mask);
        check_result(res, c_clear_cpu_mask, m_index, res);

        int stream_cpu_num = stream->get_cpu_number();

        stream->m_stream_cpus.reset();

        int threads_per_core = m_num_threads / m_num_cores;

        // The "stream_cpu_num" available threads is set in mask.
        // Available threads are defined by examining of m_cpus bitset.
        // We skip thread 0 .
        for (int i = 1; i < m_num_threads; i++) {
            // for available thread i m_cpus[i] is equal to 1
            if (m_cpus[i]) {
                res = COI::PipelineSetCPUMask(m_process,
                    i / threads_per_core,
                    i % threads_per_core,
                    in_Mask);

                check_result(res, c_set_cpu_mask, res);
                // mark thread i as nonavailable
                m_cpus.set(i,0);
                // Mark thread i as given for the stream.
                // In case of stream destroying by call to
                // _Offload_stream_destroy we can mark the thread i as
                // available.
                stream->m_stream_cpus.set(i);
                if (--stream_cpu_num <= 0) {
                    break;
                }
            }
        }

        // if stream_cpu_num is greater than 0 there are not enough
        // available threads
        if (stream_cpu_num > 0) {
            LIBOFFLOAD_ERROR(c_create_pipeline_for_stream, m_num_threads);
            LIBOFFLOAD_ABORT;
        }
        // create pipeline for this thread
        OFFLOAD_DEBUG_TRACE(2, "COIPipelineCreate Mask\n"
                               "%016lx %016lx %016lx %016lx\n%016lx %016lx %016lx %016lx\n"
                               "%016lx %016lx %016lx %016lx\n%016lx %016lx %016lx %016lx\n",
                               in_Mask[0], in_Mask[1], in_Mask[2], in_Mask[3],
                               in_Mask[4], in_Mask[5], in_Mask[6], in_Mask[7],
                               in_Mask[8], in_Mask[9], in_Mask[10], in_Mask[11],
                               in_Mask[12], in_Mask[13], in_Mask[14], in_Mask[15]);
        res = COI::PipelineCreate(m_process, in_Mask,
                                  mic_stack_size, &pipeline);
        check_result(res, c_pipeline_create, m_index, res);

        // Set stream's affinities
        {
            struct affinity_spec affinity_spec;
            char* affinity_type;
            int i;

            // "compact" by default
            affinity_spec.affinity_type = affinity_compact;

            // Check if user has specified type of affinity
            if ((affinity_type = getenv("OFFLOAD_STREAM_AFFINITY")) !=
                                        NULL)
            {
                char affinity_str[16];
                int affinity_str_len;

                OFFLOAD_DEBUG_TRACE(2,
                    "User has specified OFFLOAD_STREAM_AFFINITY=%s\n",
                    affinity_type);

                // Set type of affinity requested
                affinity_str_len = strlen(affinity_type);
                for (i=0; i<affinity_str_len && i<15; i++)
                {
                    affinity_str[i] = tolower(affinity_type[i]);
                }
                affinity_str[i] = '\0';
                if (strcmp(affinity_str, "compact") == 0) {
                    affinity_spec.affinity_type = affinity_compact;
                    OFFLOAD_DEBUG_TRACE(2, "Setting affinity=compact\n");
                } else if (strcmp(affinity_str, "scatter") == 0) {
                    affinity_spec.affinity_type = affinity_scatter;
                    OFFLOAD_DEBUG_TRACE(2, "Setting affinity=scatter\n");
                } else {
                    LIBOFFLOAD_ERROR(c_incorrect_affinity, affinity_str);
                    affinity_spec.affinity_type = affinity_compact;
                    OFFLOAD_DEBUG_TRACE(2, "Setting affinity=compact\n");
                }
            }
            // Make flat copy of sink mask because COI's mask is opaque
            for (i=0; i<16; i++) {
                affinity_spec.sink_mask[i] = in_Mask[i];
            }
            // Set number of cores and threads
            affinity_spec.num_cores = m_num_cores;
            affinity_spec.num_threads = m_num_threads;

            COIEVENT event;
            res = COI::PipelineRunFunction(pipeline,
                                   m_funcs[c_func_set_stream_affinity],
                                   0, 0, 0,
                                   0, 0,
                                   &affinity_spec, sizeof(affinity_spec),
                                   0, 0,
                                   &event);
            check_result(res, c_pipeline_run_func, m_index, res);
    
            res = COI::EventWait(1, &event, -1, 1, 0, 0);
            check_result(res, c_event_wait, res);
        }

        m_stream_lock.unlock();
        stream->set_pipeline(pipeline);
    }
    return pipeline;
}

void Engine::stream_destroy(_Offload_stream handle)
{
    // get stream
    Stream * stream =  Stream::find_stream(handle, true);

    if (stream) {
        // return cpus for future use
        for (int i = 0; i < m_num_threads; i++) {
            if (stream->m_stream_cpus.test(i)) {
                m_cpus.set(i);
            }
        }
        delete stream;
    }
    else {
        LIBOFFLOAD_ERROR(c_offload_no_stream, m_index);
        LIBOFFLOAD_ABORT;
    }
}

uint64_t Engine::get_thread_id(void)
{
    Thread* thread = (Thread*) thread_getspecific(mic_thread_key);
    if (thread == 0) {
        thread = new Thread(&m_proc_number);
        thread_setspecific(mic_thread_key, thread);
    }

    return reinterpret_cast<uint64_t>(thread);
}

AutoSet& Engine::get_auto_vars(void)
{
    Thread* thread = (Thread*) thread_getspecific(mic_thread_key);
    if (thread == 0) {
        thread = new Thread(&m_proc_number);
        thread_setspecific(mic_thread_key, thread);
    }

    return thread->get_auto_vars();
}

void Engine::destroy_thread_data(void *data)
{
    delete static_cast<Thread*>(data);
}