OpenE2K/gcc
2
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
1a8c6effec
gcc/liboffloadmic/runtime/offload_engine.cpp
Ilya Verbin 6fd2e66a09 liboffloadmic: Add missed checks for malloc and strdup return values 
liboffloadmic/
	* runtime/offload_engine.cpp (Engine::init_process): Use strdup instead
	of sizeof+malloc+sprintf, check for return value.
	* runtime/offload_env.cpp (MicEnvVar::get_env_var_kind): Check for
	strdup return value.
	* runtime/offload_host.cpp (__offload_init_library_once): Check for
	strdup return value.  Fix size calculation of COI_HOST_THREAD_AFFINITY.
	* runtime/emulator/coi_device.cpp (COIProcessWaitForShutdown): Check for
	malloc return value.

From-SVN: r228622
2015-10-08 19:04:43 +00:00

899 lines
30 KiB
C++
Raw Blame History

/*
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);
}
Reference in New Issue View Git Blame Copy Permalink