/*
* opencog/atoms/opencl/OpenclNode.cc
*
* Copyright (C) 2025 Linas Vepstas
* All Rights Reserved
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License v3 as
* published by the Free Software Foundation and including the exceptions
* at http://opencog.org/wiki/Licenses
*
* 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 Affero General Public License
* along with this program; if not, write to:
* Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*/
#include <iostream>
#include <fstream>
#include <sstream>
#include <iomanip>
#include <sys/stat.h>
#include <cstdlib>
#include <functional>
#include <opencog/util/exceptions.h>
#include <opencog/util/Logger.h>
#include <opencog/util/oc_assert.h>
#include <opencog/atomspace/AtomSpace.h>
#include <opencog/atoms/base/Node.h>
#include <opencog/atoms/value/ValueFactory.h>
#include <opencog/opencl/types/atom_types.h>
#include <opencog/sensory/types/atom_types.h>
#include "OpenclFloatValue.h"
#include "OpenclJobValue.h"
#include "OpenclNode.h"
#include "GenIDL.h"
using namespace opencog;
OpenclNode::OpenclNode(const std::string&& str) :
StreamNode(OPENCL_NODE, std::move(str)),
_dispatch_queue(this, &OpenclNode::queue_job, 1)
{
init();
}
OpenclNode::OpenclNode(Type t, const std::string&& str) :
StreamNode(t, std::move(str)),
_dispatch_queue(this, &OpenclNode::queue_job, 1)
{
if (not nameserver().isA(t, OPENCL_NODE))
throw RuntimeException(TRACE_INFO,
"Expecting OpenclNode, got %s\n", to_string().c_str());
init();
}
OpenclNode::~OpenclNode()
{
}
#define BAD_URL \
throw RuntimeException(TRACE_INFO, \
"Unsupported URL \"%s\"\n" \
"\tExpecting 'opencl://platform:device/file/path/kernel.cl'", \
url.c_str());
/// Validate the OpenCL URL
void OpenclNode::init(void)
{
const std::string& url = get_name();
if (0 != url.compare(0, 9, "opencl://")) BAD_URL;
// Ignore the first 9 chars "opencl://"
size_t pos = 9;
// Extract platform name substring
size_t platend = url.find(':', pos);
if (std::string::npos == platend) BAD_URL;
if (pos < platend)
{
_splat = url.substr(pos, platend-pos);
pos = platend;
}
pos ++;
// Extract device name substring
size_t devend = url.find('/', pos);
if (std::string::npos == devend) BAD_URL;
if (pos < devend)
{
_sdev = url.substr(pos, devend-pos);
pos = devend;
}
_filepath = url.substr(pos);
// What kind of file is it? Source or SPV?
pos = url.find_last_of('.');
if (std::string::npos == pos) BAD_URL;
_is_spv = (url.substr(pos) == ".spv");
}
// ==============================================================
void OpenclNode::find_device(void)
{
std::vector<cl::Platform> platforms;
cl::Platform::get(&platforms);
for (const auto& plat : platforms)
{
std::string pname = plat.getInfo<CL_PLATFORM_NAME>();
if (0 < _splat.size() and pname.find(_splat) == std::string::npos)
continue;
std::vector<cl::Device> devices;
plat.getDevices(CL_DEVICE_TYPE_ALL, &devices);
for (const cl::Device& dev: devices)
{
std::string dname = dev.getInfo<CL_DEVICE_NAME>();
if (dname.find(_sdev) == std::string::npos)
continue;
_platform = plat;
_device = dev;
logger().info("OpenclNode: Using platform '%s' and device '%s'\n",
pname.c_str(), dname.c_str());
return;
}
}
throw RuntimeException(TRACE_INFO,
"Unable to find platform:device in URL \"%s\"\n",
get_name().c_str());
}
// ==============================================================
// Binary caching implementation.
// Caches compiled OpenCL programs to disk to avoid expensive JIT
// compilation on subsequent runs. Cache files are stored in
// ~/.cache/opencog/opencl/<device_hash>/<source_hash>.bin
//
// This follows the pattern used by hashcat, PyOpenCL, and game engines
// to dramatically reduce startup time (from seconds to milliseconds).
/// Compute a simple hash of a string using std::hash.
/// Returns a hex string representation.
std::string OpenclNode::compute_hash(const std::string& data) const
{
std::hash<std::string> hasher;
size_t hash = hasher(data);
std::ostringstream oss;
oss << std::hex << std::setfill('0') << std::setw(16) << hash;
return oss.str();
}
/// Get the cache directory path.
/// Creates ~/.cache/opencog/opencl/<device_hash>/ if it doesn't exist.
std::string OpenclNode::get_cache_dir(void) const
{
// Get home directory
const char* home = std::getenv("HOME");
if (nullptr == home) home = "/tmp";
// Build device identifier for cache directory
std::string device_id = _platform.getInfo<CL_PLATFORM_NAME>() + "_" +
_device.getInfo<CL_DEVICE_NAME>() + "_" +
_device.getInfo<CL_DRIVER_VERSION>();
std::string device_hash = compute_hash(device_id);
std::string cache_dir = std::string(home) + "/.cache/opencog/opencl/" + device_hash;
// Create directories if they don't exist (mkdir -p equivalent)
std::string path = std::string(home) + "/.cache";
mkdir(path.c_str(), 0755);
path += "/opencog";
mkdir(path.c_str(), 0755);
path += "/opencl";
mkdir(path.c_str(), 0755);
path += "/" + device_hash;
mkdir(path.c_str(), 0755);
return cache_dir;
}
/// Get the full cache file path for a given source.
std::string OpenclNode::get_cache_path(const std::string& src) const
{
std::string source_hash = compute_hash(src);
return get_cache_dir() + "/" + source_hash + ".bin";
}
/// Try to load a cached binary. Returns true if successful.
bool OpenclNode::load_cached_binary(const std::string& cache_path)
{
std::ifstream binfile(cache_path, std::ios::binary);
if (!binfile.is_open())
return false;
// Read the entire binary file
std::vector<char> binary((std::istreambuf_iterator<char>(binfile)),
std::istreambuf_iterator<char>());
binfile.close();
if (binary.empty())
return false;
try
{
// Create program from binary
cl::Program::Binaries binaries;
binaries.push_back(std::vector<unsigned char>(binary.begin(), binary.end()));
std::vector<cl::Device> devices = {_device};
std::vector<cl_int> binary_status;
_program = cl::Program(_context, devices, binaries, &binary_status);
// Check if binary was valid for this device
if (binary_status[0] != CL_SUCCESS)
{
logger().info("OpenclNode: Cached binary invalid for device, will recompile\n");
return false;
}
// Build the program (links the binary, much faster than JIT compile)
_program.build("");
logger().info("OpenclNode: Loaded cached binary from %s\n", cache_path.c_str());
return true;
}
catch (const cl::Error& e)
{
logger().info("OpenclNode: Failed to load cached binary: %s\n", e.what());
return false;
}
}
/// Save the compiled program binary to cache.
void OpenclNode::save_binary_to_cache(const std::string& cache_path)
{
try
{
// Get the binary sizes
std::vector<size_t> binary_sizes = _program.getInfo<CL_PROGRAM_BINARY_SIZES>();
if (binary_sizes.empty() || binary_sizes[0] == 0)
{
logger().info("OpenclNode: No binary available to cache\n");
return;
}
// Get the binaries
std::vector<std::vector<unsigned char>> binaries;
binaries.resize(binary_sizes.size());
for (size_t i = 0; i < binary_sizes.size(); i++)
binaries[i].resize(binary_sizes[i]);
// Use raw pointers for the API
std::vector<unsigned char*> binary_ptrs;
for (auto& b : binaries)
binary_ptrs.push_back(b.data());
clGetProgramInfo(_program(), CL_PROGRAM_BINARIES,
binary_ptrs.size() * sizeof(unsigned char*),
binary_ptrs.data(), nullptr);
// Write to cache file
std::ofstream binfile(cache_path, std::ios::binary);
if (binfile.is_open())
{
binfile.write(reinterpret_cast<const char*>(binaries[0].data()),
binaries[0].size());
binfile.close();
logger().info("OpenclNode: Saved binary to cache: %s (%zu bytes)\n",
cache_path.c_str(), binaries[0].size());
}
}
catch (const cl::Error& e)
{
logger().info("OpenclNode: Failed to save binary to cache: %s\n", e.what());
}
}
// ==============================================================
void OpenclNode::build_program(void)
{
// Copy in source code. Must be a better way!?
std::ifstream srcfm(_filepath);
std::string src(std::istreambuf_iterator<char>(srcfm),
(std::istreambuf_iterator<char>()));
if (0 == src.size())
throw RuntimeException(TRACE_INFO,
"Unable to find source file in URL \"%s\"\n",
get_name().c_str());
// Try to load from binary cache first.
// This can reduce startup time from seconds to milliseconds.
std::string cache_path = get_cache_path(src);
bool loaded_from_cache = load_cached_binary(cache_path);
if (not loaded_from_cache)
{
// No cache hit - compile from source
logger().info("OpenclNode: Compiling kernel from source (this may take a while)...\n");
cl::Program::Sources sources;
sources.push_back(src);
_program = cl::Program(_context, sources);
// Compile
try
{
// Specifying flags causes exception.
// program.build("-cl-std=CL1.2");
_program.build("");
}
catch (const cl::Error& e)
{
logger().info("OpenclNode failed compile >>%s<<\n",
_program.getBuildInfo<CL_PROGRAM_BUILD_LOG>(_device).c_str());
throw RuntimeException(TRACE_INFO,
"Unable to compile source file in URL \"%s\"\n",
get_name().c_str());
}
// Save to cache for next time
save_binary_to_cache(cache_path);
}
// Build the interface definitions for the kernels.
// This must be done regardless of cache hit, as it creates Atomese
// for the kernel signatures.
GenIDL gidl;
HandleSeq ifcs = gidl.gen_idl(src);
HandleSeq asif;
AtomSpace *as = getAtomSpace();
for (const Handle& h : ifcs)
{
Handle has = as->add_atom(h);
_kernel_interfaces.insert({
has->getOutgoingAtom(0),
has->getOutgoingAtom(1)});
asif.emplace_back(has);
}
// Also publish them. Per sensory spec, (Predicate "*-description-*")
// is the right place. For now, stuff them into a ChoiceLink,
// because that seems to be the right thing to do. (you can only
// chose one.) They appear in the AtomSpace grouped together.
Handle descr = as->add_node(PREDICATE_NODE, "*-description-*");
Handle choice = as->add_link(CHOICE_LINK, std::move(asif));
setValue(descr, choice);
}
// ==============================================================
/// Load an SPV file.
// XXX TODO: we need some way of extracting the programming
// interfaces; not sure how to do this right now.
void OpenclNode::load_program(void)
{
// Copy in SPV file. Must be a better way!?
std::ifstream spvfm(_filepath);
std::string spv(std::istreambuf_iterator<char>(spvfm),
(std::istreambuf_iterator<char>()));
if (0 == spv.size())
throw RuntimeException(TRACE_INFO,
"Unable to find SPV file in URL \"%s\"\n",
get_name().c_str());
_program = cl::Program(_context, spv);
}
// ==============================================================
/// Attempt to open connection to OpenCL device
void OpenclNode::open(const ValuePtr& out_type)
{
if (_qvp)
throw RuntimeException(TRACE_INFO,
"Device already open! %s\n", get_name().c_str());
StreamNode::open(out_type);
if (not nameserver().isA(_item_type, FLOAT_VALUE) and
not nameserver().isA(_item_type, NUMBER_NODE))
throw RuntimeException(TRACE_INFO,
"Expecting the type to be a FloatValue or NumberNode; got %s\n",
out_type->to_string().c_str());
// Try to create the OpenCL device
find_device();
_context = cl::Context(_device);
_queue = cl::CommandQueue(_context, _device);
// Try to load source or spv file
if (_is_spv)
load_program();
else
build_program();
_qvp = createQueueValue();
}
bool OpenclNode::connected(void) const
{
return nullptr != _qvp;
}
void OpenclNode::close(const ValuePtr& ignore)
{
if (_qvp)
_qvp->close();
_qvp = nullptr;
// XXX more to do here. FIXME
// One of the TODO's is to crawl over the icoming set, look for
// OpenclKernelLinks and tell them to shut down too.
}
// ==============================================================
ValuePtr OpenclNode::stream(void) const
{
if (not connected())
throw RuntimeException(TRACE_INFO,
"Device not open! %s\n", get_name().c_str());
return _qvp;
}
ValuePtr OpenclNode::read(void) const
{
if (not connected())
throw RuntimeException(TRACE_INFO,
"Device not open! %s\n", get_name().c_str());
return _qvp->remove();
}
// ==============================================================
// This job handler runs in a different thread than the main thread.
// It uploads vectors to the GPU, and uploads and runs kernels.
// Status results are placed on the QueueValue, where the main thread
// can get at it.
//
// There are some open design questions about the sharing of queues
// and events. it might be wise to share a single `cl::Event` event
// handler, just to serialize things. If nothing else, then at least
// save to overhead of running the ctors and dtors. Same consideration
// for the queue.
void OpenclNode::queue_job(const ValuePtr& vp)
{
if (vp->is_type(OPENCL_JOB_VALUE))
{
OpenclJobValuePtr ojv = OpenclJobValueCast(vp);
// Build the kernel if not yet built. This is deferred from
// do_write() to ensure all OpenCL kernel object creation
// happens in this single dispatch thread, avoiding per-thread
// OpenCL initialization overhead.
if (not ojv->is_built())
ojv->build(ojv->get_opencl_node());
ojv->upload_inputs(get_handle());
ojv->run(get_handle());
_event_handler.wait();
_qvp->add(ojv);
return;
}
// If told to write a vector, then we upload that vector data
// to the GPU.
if (vp->is_type(OPENCL_DATA_VALUE))
{
OpenclFloatValuePtr ofv = OpenclFloatValueCast(vp);
ofv->send_buffer(get_handle());
_event_handler.wait();
_qvp->add(ofv);
return;
}
}
// ==============================================================
// Send kernel and data
void OpenclNode::write_one(const ValuePtr& kvec)
{
do_write(kvec);
}
// Prep everything needed to be able to send off a job to the GPU.
// The code here does everything that might result in an exception
// being thrown, i.e. due to user errors (e.g. badly written Atomese)
// The actual communications with the GPU is done in a distinct thread,
// so that the main thread does not hang, waiting for results to arrive.
void OpenclNode::do_write(const ValuePtr& vp)
{
// Ready-to-go. Dispatch.
if (vp->is_type(OPENCL_JOB_VALUE))
{
_dispatch_queue.enqueue(vp);
return;
}
// There's a chance that vectors haven't been attached yet.
// Do that now with set_context.
if (vp->is_type(OPENCL_DATA_VALUE))
{
OpenclFloatValuePtr ofv = OpenclFloatValueCast(vp);
ofv->set_context(get_handle());
_dispatch_queue.enqueue(vp);
return;
}
if (vp->is_type(SECTION))
{
// Create the job but DON'T build it yet. Building creates
// cl::Kernel objects which triggers OpenCL per-thread initialization.
// By deferring build() to queue_job(), all OpenCL kernel object
// creation happens in the single dispatch thread, eliminating
// the per-thread initialization overhead in CogServer.
OpenclJobValuePtr kern = createOpenclJobValue(HandleCast(vp));
kern->set_opencl_node(get_handle());
_dispatch_queue.enqueue(kern);
return;
}
throw RuntimeException(TRACE_INFO,
"Expecting data or a job, got %s\n", vp->to_string().c_str());
}
// ==============================================================
// Adds factory when library is loaded.
DEFINE_NODE_FACTORY(OpenclNode, OPENCL_NODE);
// ====================================================================
void opencog_opencl_init(void)
{
// Force shared lib ctors to run
};