aboutsummaryrefslogtreecommitdiff
path: root/fastnn
diff options
context:
space:
mode:
Diffstat (limited to 'fastnn')
-rw-r--r--fastnn/Makefile58
-rw-r--r--fastnn/device/Device.cpp496
-rw-r--r--fastnn/device/Device.h51
-rw-r--r--fastnn/device/device.c178
-rw-r--r--fastnn/device/device.lua6
-rw-r--r--fastnn/example/asgd_data_trainer.lua405
-rw-r--r--fastnn/example/asgd_sds_trainer.lua328
-rw-r--r--fastnn/fastnn-scm-1.rockspec36
-rw-r--r--fastnn/fastnn_baseline.lua236
-rw-r--r--fastnn/init.c43
-rw-r--r--fastnn/init.lua11
-rw-r--r--fastnn/io/Example.cpp186
-rw-r--r--fastnn/io/Example.h49
-rw-r--r--fastnn/io/example.c249
-rw-r--r--fastnn/io/example.lua38
-rw-r--r--fastnn/lib/ModelSync.h119
-rw-r--r--fastnn/lib/modelsync.c532
-rw-r--r--fastnn/lib/modelsync.lua107
-rw-r--r--fastnn/test.lua57
-rw-r--r--fastnn/threads/Makefile42
-rw-r--r--fastnn/threads/init.lua14
-rw-r--r--fastnn/threads/lib/THThread.c349
-rw-r--r--fastnn/threads/lib/THThread.h36
-rw-r--r--fastnn/threads/lib/init.c27
-rw-r--r--fastnn/threads/lib/luaTHRD.h84
-rw-r--r--fastnn/threads/lib/threads.c355
-rw-r--r--fastnn/threads/test/test-low-level.lua39
-rw-r--r--fastnn/threads/test/test-threads-async.lua66
-rw-r--r--fastnn/threads/test/test-threads-multiple.lua15
-rw-r--r--fastnn/threads/test/test-threads-shared.lua111
-rw-r--r--fastnn/threads/test/test-threads.lua20
-rw-r--r--fastnn/threads/threads-scm-1.rockspec36
-rw-r--r--fastnn/threads/threads.lua14
33 files changed, 4393 insertions, 0 deletions
diff --git a/fastnn/Makefile b/fastnn/Makefile
new file mode 100644
index 0000000..1f1ba1e
--- /dev/null
+++ b/fastnn/Makefile
@@ -0,0 +1,58 @@
+.PHONY: build install clean
+SHELL := /bin/bash
+BUILD_DIR := $(CURDIR)/build
+LUA_DIR = $(INST_LUADIR)/fastnn
+CUDA_BASE := /usr/local/cuda-6.5
+
+INC_PATH := -I $(LUA_BINDIR)/../include/nerv
+CUDA_INCLUDE := -I $(CUDA_BASE)/include/
+LIB_PATH := $(LUA_BINDIR)/../lib
+
+OBJ_DIR := $(BUILD_DIR)/objs
+SUBDIR := lib io threads device
+
+
+OBJS := init.o lib/modelsync.o lib/ModelSync.o io/example.o io/Example.o device/device.o device/Device.o
+LIBS := $(INST_LIBDIR)/libfastnn.so
+LUA_LIBS := init.lua lib/modelsync.lua io/example.lua device/device.lua
+INCLUDE := -I $(LUA_INCDIR) $(INC_PATH) $(CUDA_INCLUDE) -DLUA_USE_APICHECK -DUSE_PTHREAD_THREADS
+LDFLAGS := -L$(CUDA_BASE)/lib64/ -Wl,-rpath=$(CUDA_BASE)/lib64/ -lcudart -lcublas
+
+
+OBJS := $(addprefix $(OBJ_DIR)/,$(OBJS))
+OBJ_SUBDIR := $(addprefix $(OBJ_DIR)/,$(SUBDIR))
+LUA_SUBDIR := $(addprefix $(LUA_DIR)/,$(SUBDIR))
+LUA_LIBS := $(addprefix $(LUA_DIR)/,$(LUA_LIBS))
+
+CFLAGS := -Wall -Wextra -g -O0
+
+build: $(OBJ_DIR) $(OBJ_SUBDIR) $(OBJS)
+install: $(LIBS) $(LUA_DIR) $(LUA_SUBDIR) $(LUA_LIBS)
+
+#$(SUBMODULE): ECHO
+# cd threads; $(LUA_BINDIR)/luarocks make
+
+$(OBJ_DIR) $(LUA_DIR) $(OBJ_SUBDIR) $(LUA_SUBDIR):
+ -mkdir -p $@
+
+$(LUA_DIR)/%.lua: %.lua
+ cp $< $@
+
+$(OBJ_DIR)/io/Example.o: io/Example.cpp
+ gcc -c -o $@ $< $(INCLUDE) -fPIC $(CFLAGS)
+$(OBJ_DIR)/device/Device.o: device/Device.cpp
+ gcc -c -o $@ $< $(INCLUDE) -fPIC $(CFLAGS)
+
+$(OBJ_DIR)/%.o: %.c $(patsubst /%.o,/%.c,$@)
+ gcc -c -o $@ $< $(INCLUDE) -fPIC $(CFLAGS)
+
+$(LIBS): $(OBJS)
+ gcc -shared -o $@ $^ $(LDFLAGS) -Wl,-rpath=$(LIB_PATH) -L$(LIB_PATH) -lthreads -lnervcore -lluaT -lpthread -lstdc++
+
+ECHO:
+ @echo $(SUBMODULE)
+
+clean:
+ -rm -rf $(OBJ_DIR)
+
+
diff --git a/fastnn/device/Device.cpp b/fastnn/device/Device.cpp
new file mode 100644
index 0000000..3b69086
--- /dev/null
+++ b/fastnn/device/Device.cpp
@@ -0,0 +1,496 @@
+
+#include <vector>
+#include <string>
+#include <iostream>
+#include "../../nerv/lib/matrix/cuda_helper.h"
+#include <dlfcn.h>
+
+
+extern "C" {
+
+
+#include "Device.h"
+#include "stdlib.h"
+#include "string.h"
+
+
+struct GPUInfo
+{
+ int gpuid;
+ float d2h_bandwidth;
+ float h2d_bandwidth;
+ size_t mem_free;
+ size_t mem_total;
+ float mem_ratio;
+ bool used;
+};
+
+struct MPIGPUInfo
+{
+ char hostname[STRLEN];
+ int gpuid;
+ int myid;
+};
+
+struct Device
+{
+ std::vector<GPUInfo> gpuinfo_;
+ int refcount;
+};
+
+///////////////////////////////////////////
+
+extern cublasHandle_t* nerv_get_cublas_handle();
+
+CuContext* CuContext_new()
+{
+ CuContext *context = new CuContext;
+ cublasCreate(&context->cublas_handle);
+ cudaEventCreate(&context->profile_start);
+ cudaEventCreate(&context->profile_stop);
+ context->profile = hashmap_create(PROFILE_HASHMAP_SIZE, bkdr_hash, strcmp);
+ context->refcount = 1;
+ context->pid = pthread_self();
+ return context;
+}
+
+CuContext* CuContext_newWithId(long id)
+{
+ CuContext *context = (CuContext*)id;
+ __sync_fetch_and_add(&context->refcount, 1);
+ return context;
+}
+
+long CuContext_id(CuContext *context)
+{
+ return (long)context;
+}
+
+void CuContext_destroy(CuContext *context)
+{
+ if (NULL != context && __sync_fetch_and_add(&context->refcount, -1) == 1)
+ {
+ cublasDestroy(context->cublas_handle);
+ hashmap_clear(context->profile);
+ delete context;
+ context = NULL;
+ }
+}
+
+Device* Device_new()
+{
+ Device* device = new Device;
+ device->refcount = 1;
+ return device;
+}
+
+Device* Device_newWithId(long id)
+{
+ Device *device = (Device*)id;
+ __sync_fetch_and_add(&device->refcount, 1);
+ return device;
+}
+
+long Device_id(Device *device)
+{
+ return (long)device;
+}
+
+void Device_destroy(Device *device)
+{
+ if (NULL != device && __sync_fetch_and_add(&device->refcount, -1) == 1)
+ {
+ delete device;
+ device = NULL;
+ }
+}
+
+void GetFreeMemory(size_t* free, size_t* total, Status *status)
+{
+ // WARNING! the CUDA API is inconsistent accross versions!
+ #if (CUDA_VERSION >= 3020)
+ //define the function signature type
+ size_t mem_free, mem_total;
+#else
+ unsigned int mem_free, mem_total;
+#endif
+ {
+ //we will load the cuMemGetInfo dynamically from libcuda.so
+ //cuMemGetInfo(&mem_free, &mem_total);
+ //pre-fill ``safe'' values that will not cause problems
+ mem_free = 1; mem_total = 1;
+ //open libcuda.so
+ void* libcuda = dlopen("libcuda.so",RTLD_LAZY);
+ if(NULL == libcuda)
+ {
+ NERV_EXIT_STATUS(status, MAT_CUDA_ERR, "Cannot open libcuda.so");
+ }
+ else
+ {
+ //define the function signature type
+ //and get the symbol
+#if (CUDA_VERSION >= 3020)
+ typedef CUresult (*cu_fun_ptr)(size_t*, size_t*);
+ cu_fun_ptr dl_cuMemGetInfo = (cu_fun_ptr)dlsym(libcuda,"cuMemGetInfo_v2");
+#else
+ typedef CUresult (*cu_fun_ptr)(int*, int*);
+ cu_fun_ptr dl_cuMemGetInfo = (cu_fun_ptr)dlsym(libcuda,"cuMemGetInfo");
+#endif
+ if(NULL == dl_cuMemGetInfo)
+ {
+ NERV_EXIT_STATUS(status, MAT_CUDA_ERR, "Cannot load cuMemGetInfo from libcuda.so");
+ }
+ else
+ {
+ //call the function
+ dl_cuMemGetInfo(&mem_free, &mem_total);
+ }
+ //close the library
+ dlclose(libcuda);
+ }
+ }
+ // copy the output values outside
+ if(NULL != free) *free = mem_free;
+ if(NULL != total) *total = mem_total;
+}
+
+void DeviceGetName(char* name, int len, int dev, Status *status)
+{
+ //prefill with something reasonable
+ strncpy(name,"Unknown GPU",len);
+ //open libcuda.so
+ void* libcuda = dlopen("libcuda.so",RTLD_LAZY);
+ if(NULL == libcuda)
+ {
+ NERV_EXIT_STATUS(status, MAT_CUDA_ERR, "Cannot open libcuda.so");
+ }
+ else
+ {
+ //define the function signature type
+ typedef CUresult (*cu_fun_ptr)(char*,int,CUdevice);
+ //get the symbol
+ cu_fun_ptr cuDeviceGetName_ptr = (cu_fun_ptr)dlsym(libcuda,"cuDeviceGetName");
+ if(NULL == cuDeviceGetName_ptr) {
+ NERV_EXIT_STATUS(status, MAT_CUDA_ERR, "Cannot load cuDeviceGetName from libcuda.so");
+ }
+ else {
+ //call the function
+ cuDeviceGetName_ptr(name, len, dev);
+ }
+ //close the library
+ dlclose(libcuda);
+ }
+}
+
+void GetBandwidth(int gpu_idx, float *d2h, float *h2d, Status *status)
+{
+ int idx = gpu_idx;
+ int memSize = 64*1024*1024;
+ float elapsedTimeInMs = 0.0f;
+ float bandwidthInMBs = 0.0f;
+ unsigned char *h_idata = NULL;
+ unsigned char *h_odata = NULL;
+ cudaEvent_t start, stop;
+ bool PINNED = true;
+ int MEMCOPY_ITERATIONS = 5;
+
+ CUDA_SAFE_SYNC_CALL(cudaEventCreate(&start), status);
+ CUDA_SAFE_SYNC_CALL(cudaEventCreate(&stop), status);
+
+ //allocate host memory
+ if (PINNED)
+ {
+ #if CUDART_VERSION >= 2020
+ CUDA_SAFE_SYNC_CALL(cudaHostAlloc((void **)&h_idata, memSize, cudaHostAllocPortable), status);
+ CUDA_SAFE_SYNC_CALL(cudaHostAlloc((void **)&h_odata, memSize, cudaHostAllocPortable), status);
+ #else
+ CUDA_SAFE_SYNC_CALL(cudaMallocHost((void **)&h_idata, memSize), status);
+ CUDA_SAFE_SYNC_CALL(cudaMallocHost((void **)&h_odata, memSize), status);
+ #endif
+ }
+ else
+ {
+ //pageable memory mode - use malloc
+ h_odata = (unsigned char *)malloc(memSize);
+
+ if (h_odata == 0)
+ {
+ NERV_EXIT_STATUS(status, MAT_CUDA_ERR, "Not enough memory available on host to run test!");
+ }
+ }
+
+ //initialize the memory
+ for (unsigned int i = 0; i < memSize/sizeof(unsigned char); i++)
+ {
+ h_idata[i] = (unsigned char)(i & 0xff);
+ }
+
+ // allocate device memory
+ unsigned char *d_idata;
+ CUDA_SAFE_SYNC_CALL(cudaMalloc((void **) &d_idata, memSize), status);
+
+ //initialize the device memory
+ CUDA_SAFE_SYNC_CALL(cudaMemcpy(d_idata, h_idata, memSize,cudaMemcpyHostToDevice), status);
+
+ //copy data from GPU to Host
+
+ CUDA_SAFE_SYNC_CALL(cudaEventRecord(start, 0), status);
+
+ if (PINNED)
+ {
+ for (unsigned int i = 0; i < MEMCOPY_ITERATIONS; i++)
+ {
+ CUDA_SAFE_SYNC_CALL(cudaMemcpyAsync(h_odata, d_idata, memSize,cudaMemcpyDeviceToHost, 0), status);
+ }
+ }
+ else
+ {
+ for (unsigned int i = 0; i < MEMCOPY_ITERATIONS; i++)
+ {
+ CUDA_SAFE_SYNC_CALL(cudaMemcpy(h_odata, d_idata, memSize,cudaMemcpyDeviceToHost), status);
+ }
+ }
+
+ CUDA_SAFE_SYNC_CALL(cudaEventRecord(stop, 0), status);
+
+ // make sure GPU has finished copying
+ CUDA_SAFE_SYNC_CALL(cudaDeviceSynchronize(), status);
+ //get the the total elapsed time in ms
+
+ CUDA_SAFE_SYNC_CALL(cudaEventElapsedTime(&elapsedTimeInMs, start, stop), status);
+
+
+ //calculate bandwidth in MB/s
+ bandwidthInMBs = (1e3f * memSize * (float)MEMCOPY_ITERATIONS) / (elapsedTimeInMs * (float)(1 << 20));
+ *d2h = bandwidthInMBs;
+
+ /////////////////////////////////////////////////////
+ //copy data from Host to GPU
+ CUDA_SAFE_SYNC_CALL(cudaEventRecord(start, 0), status);
+
+ if (PINNED)
+ {
+ for (unsigned int i = 0; i < MEMCOPY_ITERATIONS; i++)
+ {
+ CUDA_SAFE_SYNC_CALL(cudaMemcpyAsync(d_idata, h_odata, memSize,cudaMemcpyHostToDevice, 0), status);
+ }
+ }
+ else
+ {
+ for (unsigned int i = 0; i < MEMCOPY_ITERATIONS; i++)
+ {
+ CUDA_SAFE_SYNC_CALL(cudaMemcpy(d_idata, h_odata, memSize,cudaMemcpyHostToDevice), status);
+ }
+ }
+
+ CUDA_SAFE_SYNC_CALL(cudaEventRecord(stop, 0), status);
+
+ // make sure GPU has finished copying
+ CUDA_SAFE_SYNC_CALL(cudaDeviceSynchronize(), status);
+ //get the the total elapsed time in ms
+
+ CUDA_SAFE_SYNC_CALL(cudaEventElapsedTime(&elapsedTimeInMs, start, stop), status);
+
+
+ //calculate bandwidth in MB/s
+ bandwidthInMBs = (1e3f * memSize * (float)MEMCOPY_ITERATIONS) / (elapsedTimeInMs * (float)(1 << 20));
+ *h2d = bandwidthInMBs;
+
+ //clean up memory
+ CUDA_SAFE_SYNC_CALL(cudaEventDestroy(stop), status);
+ CUDA_SAFE_SYNC_CALL(cudaEventDestroy(start), status);
+
+ if (PINNED)
+ {
+ CUDA_SAFE_SYNC_CALL(cudaFreeHost(h_idata), status);
+ CUDA_SAFE_SYNC_CALL(cudaFreeHost(h_odata), status);
+ }
+ else
+ {
+ free(h_idata);
+ free(h_odata);
+ }
+
+ CUDA_SAFE_SYNC_CALL(cudaFree(d_idata), status);
+}
+
+int AutoSelectGPU(Device *device, Status *status)
+{
+ //GPU selection is based on largest proportion of free memory.
+
+ int max_id = 0, n_gpu = 0;
+ std::vector<GPUInfo> &gpuinfo_ = device->gpuinfo_;
+
+ cudaGetDeviceCount(&n_gpu);
+
+ if(n_gpu == 0)
+ {
+ NERV_SET_STATUS(status, MAT_CUDA_ERR, "No CUDA devices found");
+ return -1;
+ }
+
+ if (n_gpu > 0)
+ {
+ for (int i = 0; i < gpuinfo_.size(); i++)
+ {
+ if (!gpuinfo_[i].used)
+ {
+ max_id = i;
+ break;
+ }
+ }
+ //find GPU with max free memory
+ for (int n = 1; n < gpuinfo_.size(); n++)
+ {
+ if (!gpuinfo_[n].used && gpuinfo_[n].mem_ratio > gpuinfo_[max_id].mem_ratio)
+ max_id = n;
+ }
+
+
+ std::cerr << "Selected device: " << max_id << " (automatically)" << std::endl;
+
+ std::cerr << "free: " << gpuinfo_[max_id].mem_free/1024/1024 << "M, "
+ << "total: "<< gpuinfo_[max_id].mem_total/1024/1024 << "M, "
+ << "ratio: "<< gpuinfo_[max_id].mem_ratio << " "
+ << "d2h bandwidth: " << gpuinfo_[max_id].d2h_bandwidth << "MB/s, "
+ << "h2d bandwidth: "<< gpuinfo_[max_id].h2d_bandwidth << "MB/s" << std::endl;
+
+ cudaSetDevice(max_id);
+ //initialize the CUBLAS
+ //cublasInit();
+ //cublasHandle_t *cublas_handle = nerv_get_cublas_handle();
+ //std::cerr << "cublasHandle_t: " << cublas_handle << std::endl;
+ //cublasCreate(cublas_handle);
+
+ //create the context
+ cudaError_t e;
+ e = cudaThreadSynchronize(); //deprecated, but for legacy not cudaDeviceSynchronize
+ if(e != cudaSuccess) {
+ std::cerr << "Failed to create CUDA context on a GPU." << std::endl;
+ }
+ }
+
+ gpuinfo_[max_id].used = true;
+ NERV_SET_STATUS(status, NERV_NORMAL, 0);
+ return max_id;
+}
+
+void SelectGPU(Device *device, int gpu_id, Status *status)
+{
+
+ std::vector<GPUInfo> &gpuinfo_ = device->gpuinfo_;
+ int n_gpu = 0;
+ cudaGetDeviceCount(&n_gpu);
+ if(gpu_id >= n_gpu)
+ {
+ NERV_EXIT_STATUS(status, MAT_CUDA_ERR, "Cannot select GPU CUDA capable cards!");
+ std::cerr << "Cannot select GPU " << gpu_id
+ << ", detected " << n_gpu << " CUDA capable cards!" << std::endl;
+ }
+
+
+ std::cerr << "Selected device: " << gpu_id << " (manually)";
+
+ std::cerr << "free: " << gpuinfo_[gpu_id].mem_free/1024/1024 << "M, "
+ << "total: "<< gpuinfo_[gpu_id].mem_total/1024/1024 << "M, "
+ << "ratio: "<< gpuinfo_[gpu_id].mem_ratio << " "
+ << "d2h bandwidth: " << gpuinfo_[gpu_id].d2h_bandwidth << "MB/s, "
+ << "h2d bandwidth: "<< gpuinfo_[gpu_id].h2d_bandwidth << "MB/s" << std::endl;
+
+ CUDA_SAFE_SYNC_CALL(cudaSetDevice(gpu_id), status);
+ //initialize the CUBLAS
+ //CUBLAS_SAFE_SYNC_CALL(cublasInit(), status);
+ //cublasHandle_t *cublas_handle = nerv_get_cublas_handle();
+ //cublasCreate(cublas_handle);
+
+ //create the context
+ cudaError_t e;
+ e = cudaThreadSynchronize(); //deprecated, but for legacy not cudaDeviceSynchronize
+ if(e != cudaSuccess) {
+ NERV_EXIT_STATUS(status, MAT_CUDA_ERR, "Failed to create CUDA context on a GPU.");
+ std::cerr << "Failed to create CUDA context on a GPU.";
+ }
+
+ gpuinfo_[gpu_id].used = true;
+
+ NERV_SET_STATUS(status, NERV_NORMAL, 0);
+}
+
+void Initialize(Device *device, Status *status)
+{
+ // Check that we have at least one gpu
+ int n_gpu = 0;
+ cudaGetDeviceCount(&n_gpu);
+ if(n_gpu == 0)
+ NERV_EXIT_STATUS(status, MAT_CUDA_ERR, "No CUDA devices found");
+
+ std::vector<GPUInfo> &gpuinfo_ = device->gpuinfo_;
+ gpuinfo_.resize(n_gpu);
+
+ std::cerr << "gpu information ..." << std::endl;
+
+ // Get ratios of memory use, if possible
+ for(int n = 0; n < n_gpu; n++)
+ {
+ int ret = cudaSetDevice(n);
+ switch(ret)
+ {
+ case cudaSuccess : {
+ //create the CUDA context for the thread
+ //cudaThreadSynchronize(); //deprecated, but for legacy not cudaDeviceSynchronize
+ cudaDeviceSynchronize();
+ //get GPU name
+ char name[STRLEN];
+ DeviceGetName(name,STRLEN,n, status);
+ //get GPU memory stats
+ size_t free, total;
+ float d2h, h2d;
+
+ GetFreeMemory(&free, &total, status);
+ GetBandwidth(n, &d2h, &h2d, status);
+
+ gpuinfo_[n].gpuid = n;
+ gpuinfo_[n].d2h_bandwidth = d2h;
+ gpuinfo_[n].h2d_bandwidth = h2d;
+ gpuinfo_[n].mem_free = free;
+ gpuinfo_[n].mem_total = total;
+ gpuinfo_[n].mem_ratio = free/(float)total;
+ gpuinfo_[n].used = false;
+
+ std::cerr << "gpu: " << n << " ==> "
+ << "free: " << free/1024/1024 << "M, "
+ << "total: "<< total/1024/1024 << "M, "
+ << "ratio: "<< free/(float)total << " "
+ << "d2h bandwidth: " << d2h << "MB/s, "
+ << "h2d bandwidth: "<< h2d << "MB/s" << std::endl;
+
+ //destroy the CUDA context for the thread
+ //cudaThreadExit(); //deprecated, but for legacy reason not cudaDeviceReset
+ } break;
+
+ #if (CUDA_VERSION > 3020)
+ case cudaErrorDeviceAlreadyInUse :
+ std::cerr << "cudaSetDevice(" << n << "): "
+ << "Device cannot be accessed, used EXCLUSIVE-THREAD mode..." << std::endl;
+ break;
+ #endif
+ case cudaErrorInvalidDevice :
+ std::cerr << "cudaSetDevice(" << n << "): "
+ << "Device cannot be accessed, not a VALID CUDA device!" << std::endl;
+ break;
+ default :
+ std::cerr << "cudaSetDevice(" << n << "): "
+ << "returned " << ret << ", "
+ << cudaGetErrorString((cudaError_t)ret) << std::endl;
+ }
+ }
+ cudaDeviceReset();
+ printf("Result = PASS\n");
+ NERV_SET_STATUS(status, NERV_NORMAL, 0);
+}
+
+
+
+
+
+} // extern
diff --git a/fastnn/device/Device.h b/fastnn/device/Device.h
new file mode 100644
index 0000000..9a80128
--- /dev/null
+++ b/fastnn/device/Device.h
@@ -0,0 +1,51 @@
+
+#ifndef NERV_FASTNN_EXAMPLE_H
+#define NERV_FASTNN_EXAMPLE_H
+
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "matrix/matrix.h"
+#include "stdbool.h"
+#include "../../nerv/lib/matrix/cuda_helper.h"
+#include "../../nerv/lib/common.h"
+#include "../../nerv/lib/matrix/cumatrix.h"
+#define STRLEN 1024
+
+
+
+typedef struct GPUInfo GPUInfo;
+
+typedef struct MPIGPUInfo MPIGPUInfo;
+
+typedef struct Device Device;
+
+CuContext* CuContext_new();
+CuContext* CuContext_newWithId(long id);
+long CuContext_id(CuContext *context);
+void CuContext_destroy(CuContext *context);
+
+
+Device* Device_new();
+Device* Device_newWithId(long);
+long Device_id(Device *device);
+void Device_detroy(Device *device);
+void Initialize(Device *device, Status *status);
+void GetFreeMemory(size_t* free, size_t* total, Status *status);
+void DeviceGetName(char* name, int len, int dev, Status *status);
+void GetBandwidth(int gpu_idx, float *d2h, float *h2d, Status *status);
+int AutoSelectGPU(Device *device, Status *status);
+void SelectGPU(Device *device, int gpu_id, Status *status);
+
+
+
+
+#ifdef __cplusplus
+} // closing brace for extern "C"
+
+#endif // end Device
+
+#endif
+
diff --git a/fastnn/device/device.c b/fastnn/device/device.c
new file mode 100644
index 0000000..71d6ec1
--- /dev/null
+++ b/fastnn/device/device.c
@@ -0,0 +1,178 @@
+
+#include <stdio.h>
+#include <stdlib.h>
+
+#include <lua.h>
+#include <lualib.h>
+#include <luaT/luaT.h>
+
+#include "../threads/lib/luaTHRD.h"
+#include "Device.h"
+
+
+const char *fastnn_device_tname = "fastnn.CDevice";
+
+
+static int device_new(lua_State *L)
+{
+ Device *device = NULL;
+ if(lua_gettop(L) == 0)
+ {
+ device = Device_new();
+ }
+ else if(lua_gettop(L) == 1)
+ {
+ long id = luaL_checkinteger(L, 1);
+ device = Device_newWithId(id);
+ }
+ else
+ luaL_error(L, "device: device new invalid arguments");
+ if (!device)
+ luaL_error(L, "device: device failed");
+
+ luaTHRD_pushudata(L, device, fastnn_device_tname);
+
+ return 1;
+}
+
+static int device_init(lua_State *L)
+{
+ Device *device = luaTHRD_checkudata(L, 1, fastnn_device_tname);
+ Status status;
+ Initialize(device, &status);
+ NERV_LUA_CHECK_STATUS(L, status);
+ return 0;
+}
+
+static int device_select_gpu(lua_State *L)
+{
+ Device *device = luaTHRD_checkudata(L, 1, fastnn_device_tname);
+
+ Status status;
+ if(lua_gettop(L) == 2)
+ {
+ int gpuid = luaL_checkinteger(L, 2);
+ SelectGPU(device, gpuid, &status);
+ NERV_LUA_CHECK_STATUS(L, status);
+ return 0;
+ }
+ else if(lua_gettop(L) == 1)
+ {
+ int gpuid = AutoSelectGPU(device, &status);
+ NERV_LUA_CHECK_STATUS(L, status);
+ lua_pushinteger(L, gpuid);
+ return 1;
+ }
+ else
+ luaL_error(L, "device: device select gpu failed");
+}
+
+static int device_id(lua_State *L)
+{
+ Device *device = luaTHRD_checkudata(L, 1, fastnn_device_tname);
+ lua_pushinteger(L, Device_id(device));
+ return 1;
+}
+
+static int device_tostring(lua_State *L)
+{
+ char str[STRLEN];
+ Device* device = luaTHRD_checkudata(L, 1, fastnn_device_tname);
+ snprintf(str, STRLEN, "%s <%lx>", fastnn_device_tname, Device_id(device));
+ lua_pushstring(L, str);
+ return 1;
+}
+
+static int device_destroy(lua_State *L)
+{
+ Device *device = luaTHRD_checkudata(L, 1, fastnn_device_tname);
+ Device_destroy(device);
+ //printf("device_destroy ... end\n");
+ return 0;
+}
+
+
+//////////////////////////////////////////////
+
+static int context_new(lua_State *L)
+{
+ CuContext *context = NULL;
+ if(lua_gettop(L) == 0)
+ {
+ context = CuContext_new();
+ }
+ else if(lua_gettop(L) == 1)
+ {
+ long id = luaL_checkinteger(L, 1);
+ context = CuContext_newWithId(id);
+ }
+ else
+ luaL_error(L, "device: context new invalid arguments");
+ if (!context)
+ luaL_error(L, "device: context failed");
+
+ luaTHRD_pushudata(L, context, nerv_context_tname);
+
+ return 1;
+}
+
+
+static int context_id(lua_State *L)
+{
+ CuContext *context = luaTHRD_checkudata(L, 1, nerv_context_tname);
+ lua_pushinteger(L, CuContext_id(context));
+ return 1;
+}
+
+static int context_tostring(lua_State *L)
+{
+ char str[STRLEN];
+ CuContext* context = luaTHRD_checkudata(L, 1, nerv_context_tname);
+ snprintf(str, STRLEN, "%s <%lx>", nerv_context_tname, CuContext_id(context));
+ lua_pushstring(L, str);
+ return 1;
+}
+
+static int context_destroy(lua_State *L)
+{
+ CuContext* context = luaTHRD_checkudata(L, 1, nerv_context_tname);
+ CuContext_destroy(context);
+ return 0;
+}
+
+
+static const struct luaL_Reg device__ [] = {
+ {"new", device_new},
+ {"__tostring", device_tostring},
+ {"id", device_id},
+ {"init", device_init},
+ {"select_gpu", device_select_gpu},
+ {"free", device_destroy},
+ {NULL, NULL}
+};
+
+static const struct luaL_Reg context__ [] = {
+ {"new", context_new},
+ {"__tostring", context_tostring},
+ {"id", context_id},
+ {"free", context_destroy},
+ {NULL, NULL}
+};
+
+
+void fastnn_init_device(lua_State *L)
+{
+ luaT_newmetatable(L, fastnn_device_tname, NULL, device_new, device_destroy, NULL);
+ luaL_register(L, NULL, device__);
+ lua_pop(L, 1);
+}
+
+void fastnn_init_context(lua_State *L)
+{
+ luaT_newmetatable(L, nerv_context_tname, NULL, context_new, context_destroy, NULL);
+ luaL_register(L, NULL, context__);
+ lua_pop(L, 1);
+}
+
+
+
diff --git a/fastnn/device/device.lua b/fastnn/device/device.lua
new file mode 100644
index 0000000..d3dea73
--- /dev/null
+++ b/fastnn/device/device.lua
@@ -0,0 +1,6 @@
+
+local C = require 'libfastnn'
+
+fastnn.CDevice = C.CDevice
+
+
diff --git a/fastnn/example/asgd_data_trainer.lua b/fastnn/example/asgd_data_trainer.lua
new file mode 100644
index 0000000..33d579a
--- /dev/null
+++ b/fastnn/example/asgd_data_trainer.lua
@@ -0,0 +1,405 @@
+require 'fastnn'
+require 'libhtkio'
+require 'threads'
+
+dofile("fastnn/fastnn_baseline.lua")
+
+env = string.format([[
+package.path="/home/slhome/wd007/.luarocks/share/lua/5.1/?.lua;/home/slhome/wd007/.luarocks/share/lua/5.1/?/init.lua;/sgfs/users/wd007/src/nerv/install/share/lua/5.1/?.lua;/sgfs/users/wd007/src/nerv/install/share/lua/5.1/?/init.lua;"..package.path;
+package.cpath="/home/slhome/wd007/.luarocks/lib/lua/5.1/?.so;/sgfs/users/wd007/src/nerv/install/lib/lua/5.1/?.so;"..package.cpath
+local k,l,_=pcall(require,"luarocks.loader") _=k and l.add_context("nerv","scm-1")
+]])
+
+
+
+local data_thread_code = [[
+%s
+
+require 'nerv'
+require 'fastnn'
+dofile("fastnn/fastnn_baseline.lua")
+os.execute("export MALLOC_CHECK_=0")
+
+local thread_idx = %d
+local example_repo_shareid = %d
+local data_mutex_shareid = %d
+local feat_repo_shareid = %d
+local gpu_shareid = %d
+local batch_size = %d
+local bp = %d
+local scp_file = '%s'
+
+local share_mutex = threads.Mutex(data_mutex_shareid)
+local share_example_repo = fastnn.CExamplesRepo(example_repo_shareid, true)
+local share_gpu = fastnn.CDevice(gpu_shareid)
+
+--print(thread_idx)
+--print(share_mutex)
+--print(share_gpu)
+--print(share_example_repo)
+
+if bp == 0 then
+ bp = false
+else
+ bp = true
+end
+gconf.randomize = bp
+--print(gconf.randomize)
+
+share_mutex:lock()
+local gpuid = share_example_repo:get_gpuid()
+if gpuid < 0 then
+ gpuid = share_gpu:select_gpu()
+ share_example_repo:set_gpuid(gpuid)
+else
+ share_gpu:select_gpu(gpuid)
+end
+
+nerv.info_stderr("thread %%d loading transf ...", thread_idx)
+local param_transf_repo = nerv.ParamRepo()
+param_transf_repo:import(gconf.transf, nil, gconf)
+local transf_node_repo = make_transf_node_repo(param_transf_repo)
+local transf_layer_repo = make_transf_link_repo(transf_node_repo, param_transf_repo)
+local transf = transf_layer_repo:get_layer("global_transf")
+share_mutex:unlock()
+
+local feat_id = get_feat_id()
+
+local buffer = make_buffer(make_readers(scp_file, transf_layer_repo, feat_repo_shareid, data_mutex_shareid))
+
+ local t = 1;
+ for data in buffer.get_data, buffer do
+ local example = fastnn.Example:PrepareData(data, nil, feat_id)
+ --print(string.format("Accept NO.%%d %%s", t, example)); t = t+1;
+ share_example_repo:accept(example)
+ --print("share_example_repo:accept")
+
+ -- collect garbage in-time to save GPU memory
+ collectgarbage("collect")
+ end
+ share_example_repo:done()
+-- print("share_example_repo:done")
+
+]]
+
+
+train_thread_code = [[
+%s
+
+require 'nerv'
+require 'fastnn'
+dofile("fastnn/fastnn_baseline.lua")
+os.execute("export MALLOC_CHECK_=0")
+
+local thread_idx = %d
+local example_repo_shareid = %d
+local data_mutex_shareid = %d
+local master_shareid = %d
+local gpu_shareid = %d
+local xent_shareid = %d
+local batch_size = %d
+local lrate = %f
+local bp = %d
+local nnet_in = '%s'
+local nnet_out = '%s'
+
+local share_example_repo = fastnn.CExamplesRepo(example_repo_shareid, true)
+local share_mutex = threads.Mutex(data_mutex_shareid)
+local share_master = fastnn.ModelSync(master_shareid)
+local share_gpu = fastnn.CDevice(gpu_shareid)
+local share_xent = fastnn.CXent(xent_shareid)
+
+if bp == 0 then
+ bp = false
+else
+ bp = true
+end
+
+gconf.randomize = bp
+gconf.lrate = lrate
+gconf.batch_size = batch_size
+gconf.network[1] = nnet_in
+nerv.info_stderr("input network: %%s", gconf.network[1])
+nerv.info_stderr(gconf.randomize)
+nerv.info_stderr("input batch_size: %%d", gconf.batch_size)
+nerv.info_stderr("input lrate: %%f", gconf.lrate)
+
+share_mutex:lock()
+local gpuid = share_example_repo:get_gpuid()
+if gpuid < 0 then
+ gpuid = share_gpu:select_gpu()
+ share_example_repo:set_gpuid(gpuid)
+else
+ share_gpu:select_gpu(gpuid)
+end
+
+nerv.context = nerv.CCuContext()
+--print(nerv.context)
+
+
+nerv.info_stderr("thread %%d loading network ...", thread_idx)
+local param_network_repo = nerv.ParamRepo()
+param_network_repo:import(gconf.network, nil, gconf)
+local network_node_repo = make_network_node_repo(param_network_repo)
+local network_layer_repo = make_network_link_repo(network_node_repo, param_network_repo)
+local network = get_network(network_layer_repo)
+share_mutex:unlock()
+
+
+ local input_order = get_input_order()
+
+ -- initialize the network
+ network:init(gconf.batch_size)
+ gconf.cnt = 0
+ err_input = {nerv.CuMatrixFloat(gconf.batch_size, 1)}
+ err_input[1]:fill(1)
+
+ share_master:Initialize(network)
+ share_master:SyncInc()
+
+ for example in share_example_repo.provide, share_example_repo do
+
+ gconf.cnt = gconf.cnt + 1
+ if gconf.cnt == 2000 then
+ print_stat(network_node_repo)
+ gconf.cnt = 0
+ end
+
+ local input = {}
+ local n = example:size()
+ for i = 0, n-1 do
+ table.insert(input, example:at(i))
+ end
+
+ local output = {nerv.CuMatrixFloat(gconf.batch_size, 1)}
+ err_output = {input[1]:create()}
+ network:propagate(input, output)
+
+ if bp then
+ network:back_propagate(err_input, err_output, input, output)
+ network:gradient(err_input, input, output)
+
+ share_master:LockModel()
+ share_master:WeightToD(network)
+ network:update_gradient()
+ share_master:WeightFromD(network)
+ share_master:UnLockModel()
+ end
+
+ -- collect garbage in-time to save GPU memory
+ collectgarbage("collect")
+ end
+
+ --print_stat(network_node_repo)
+ local ce_crit = network_node_repo:get_layer("ce_crit")
+ local xent = fastnn.CXent(ce_crit.total_frames, ce_crit.total_correct, ce_crit.total_ce, ce_crit.total_ce)
+
+ share_master:LockModel()
+ share_xent:add(xent)
+ share_master:SyncDec()
+ --print(string.format("ThreadCount: %%d", share_master:ThreadCount()))
+ if share_master:ThreadCount() == 0 and bp then
+ share_master:WeightToD(network)
+ local fname = string.format("%%s_tr%%.3f",
+ nnet_out, frame_acc(share_xent))
+ nerv.info_stderr("writing back %%s ...", fname)
+ network:get_params():export(fname, nil)
+ end
+ share_master:UnLockModel()
+]]
+
+function get_data_thread(data_thread_code, env, thread_idx, example_repo_shareid,
+ data_mutex_shareid, feat_repo_shareid, gpu_shareid,
+ batch_size, bp, scp_file)
+ return string.format(data_thread_code, env, thread_idx, example_repo_shareid,
+ data_mutex_shareid, feat_repo_shareid, gpu_shareid,
+ batch_size, bp, scp_file)
+end
+
+function get_train_thread(train_thread_code, env, thread_idx, example_repo_shareid,
+ data_mutex_shareid, master_shareid, gpu_shareid, xent_shareid,
+ batch_size, lrate, bp, nnet_in, nnet_out)
+ return string.format(train_thread_code, env, thread_idx, example_repo_shareid,
+ data_mutex_shareid, master_shareid, gpu_shareid, xent_shareid,
+ batch_size, lrate, bp, nnet_in, nnet_out)
+end
+
+function trainer(batch_size, lrate, bp, scp_file, nnet_in, nnet_out, num_threads)
+ local train_threads={}
+ local trainer = {}
+ local data_threads = {}
+ local data = {}
+ local num_threads=num_threads
+
+ local data_mutex = threads.Mutex()
+ local data_mutex_shareid = data_mutex:id()
+
+ local master = fastnn.CModelSync()
+ local master_shareid = master:id()
+ --print(master)
+
+ local xent = fastnn.CXent()
+ local xent_shareid = xent:id()
+ --print(xent)
+
+ local gpu = fastnn.CDevice()
+ local gpu_shareid = gpu:id()
+ --print(gpu_shareid)
+ gpu:init()
+
+ local example_repo = {}
+ local example_repo_shareid = {}
+
+ local feat_repo = nerv.TNetFeatureRepo(scp_file, gconf.htk_conf, gconf.frm_ext)
+ local feat_repo_shareid = feat_repo:id()
+
+ for i=1,num_threads,1 do
+ example_repo[i] = fastnn.CExamplesRepo(128, false)
+ example_repo_shareid[i] = example_repo[i]:id()
+
+ data_threads[i] = get_data_thread(data_thread_code, env, i, example_repo_shareid[i],
+ data_mutex_shareid, feat_repo_shareid, gpu_shareid,
+ batch_size, bp, scp_file)
+
+ train_threads[i] = get_train_thread(train_thread_code, env, i, example_repo_shareid[i],
+ data_mutex_shareid, master_shareid, gpu_shareid, xent_shareid,
+ batch_size, lrate, bp, nnet_in, nnet_out)
+ --print(train_threads[i])
+ data[i] = threads.Thread(data_threads[i])
+ trainer[i] = threads.Thread(train_threads[i])
+ end
+
+ nerv.info_stderr('| waiting for thread...')
+
+ for i=1,num_threads,1 do
+ data[i]:free()
+ trainer[i]:free()
+ end
+
+ print_xent(xent)
+
+ nerv.info_stderr('| all thread finished!')
+
+ return frame_acc(xent)
+end
+
+function get_filename(fname)
+ return string.gsub((string.gsub(fname, "(.*/)(.*)", "%2")),"(.*)%..*", "%1")
+end
+
+function do_sds(tr_scp, sds_scp, sds_rate)
+ math.randomseed(os.time())
+ local scp_file = io.open(tr_scp, "r")
+ local sds_file = io.open(sds_scp, "w")
+ for line in scp_file:lines() do
+ rate = math.random()
+ if (rate < sds_rate) then
+ sds_file:write(line.."\n")
+ end
+ end
+ scp_file:close()
+ sds_file:close()
+end
+
+function print_tag(iter)
+ io.stderr:write(string.format("########################################################\n"))
+ io.stderr:write(string.format("# NN TRAINING ITERATION:%d, %s\n", iter, os.date()))
+ io.stderr:write(string.format("########################################################\n"))
+end
+
+
+start_halving_inc = 0.5
+halving_factor = 0.8
+end_halving_inc = 0.1
+min_iter = 1
+max_iter = 20
+min_halving = 0
+gconf.batch_size = 256
+pf0 = get_filename(gconf.network[1])
+nnet_in = gconf.network[1]
+nnet_out = ""
+sds_scp = "tr_sds_"..string.format("%.4d", math.random()*10000)..".scp" --"tr_sds.scp"
+sds_factor = 0.4
+num_threads = 1
+global_option = nil
+
+print_gconf()
+os.execute("export MALLOC_CHECK_=0")
+
+-- training begin
+nerv.info_stderr("begin initial cross validation")
+local accu_best = trainer(gconf.batch_size, gconf.lrate, 0,
+ gconf.cv_scp, nnet_in, nil, num_threads)
+local do_halving = false
+local accu_new = accu_best
+
+nerv.info_stderr("initial cross validation: %.3f\n", accu_best)
+
+for i = 1, max_iter do
+
+ if accu_new >= accu_best then
+ local sds_rate = math.cos((i-1)*11.0/180*math.pi)
+ if (sds_rate <= sds_factor) then
+ sds_rate = sds_factor
+ end
+ nerv.info_stderr("iteration %d sds_rate: %.6f", i, sds_rate)
+ do_sds(gconf.tr_scp, sds_scp, sds_rate)
+ end
+
+ nnet_out=pf0.."_iter"..i
+ --print(nnet_out)
+ print_tag(i)
+ nerv.info_stderr("[NN] begin iteration %d learning_rate: %.3f batch_size: %d.", i, gconf.lrate, gconf.batch_size)
+ local accu_tr = trainer(gconf.batch_size, gconf.lrate, 1,
+ sds_scp, nnet_in, nnet_out, num_threads)
+ nerv.info_stderr("[TR] end iteration %d frame_accuracy: %.3f.\n", i, accu_tr)
+ os.execute("sleep " .. 3)
+
+ nnet_out = nnet_out.."_tr"..accu_tr
+ accu_new = trainer(gconf.batch_size, gconf.lrate, 0,
+ gconf.cv_scp, nnet_out, nil, num_threads)
+ nerv.info_stderr("[CV] end iteration %d frame_accuracy: %.3f.\n\n", i, accu_new)
+ os.execute("sleep " .. 3)
+
+ local nnet_tmp = string.format("%s_%s_iter_%d_lr%f_tr%.3f_cv%.3f",
+ pf0,
+ os.date("%Y%m%d%H%M%S"),
+ i, gconf.lrate, accu_tr, accu_new)
+
+ -- TODO: revert the weights
+ local accu_diff = accu_new - accu_best
+ local cmd
+ if accu_new > accu_best then
+ accu_best = accu_new
+ nnet_in = nnet_tmp
+ gconf.batch_size = gconf.batch_size + 128
+ if gconf.batch_size > 1024 then
+ gconf.batch_size = 1024
+ end
+ else
+ -- reject
+ nnet_tmp = nnet_tmp.."_rejected"
+ do_halving = true
+ end
+ cmd = "mv "..nnet_out.." "..nnet_tmp
+ os.execute(cmd)
+
+ if do_halving and accu_diff < end_halving_inc and i > min_iter then
+ break;
+ end
+
+ if accu_diff < start_halving_inc and i >= min_halving then
+ do_halving = true
+ end
+
+ if do_halving then
+ gconf.lrate = gconf.lrate * halving_factor
+ halving_factor = halving_factor - 0.025
+ if halving_factor < 0.6 then
+ halving_factor = 0.6
+ end
+ end
+ nerv.info_stderr("iteration %d done!", i)
+end
+
+
diff --git a/fastnn/example/asgd_sds_trainer.lua b/fastnn/example/asgd_sds_trainer.lua
new file mode 100644
index 0000000..44611b2
--- /dev/null
+++ b/fastnn/example/asgd_sds_trainer.lua
@@ -0,0 +1,328 @@
+package.path="/home/slhome/wd007/.luarocks/share/lua/5.1/?.lua;/home/slhome/wd007/.luarocks/share/lua/5.1/?/init.lua;/sgfs/users/wd007/src/nerv/install/share/lua/5.1/?.lua;/sgfs/users/wd007/src/nerv/install/share/lua/5.1/?/init.lua;"..package.path;
+package.cpath="/home/slhome/wd007/.luarocks/lib/lua/5.1/?.so;/sgfs/users/wd007/src/nerv/install/lib/lua/5.1/?.so;"..package.cpath
+local k,l,_=pcall(require,"luarocks.loader") _=k and l.add_context("nerv","scm-1")
+require 'nerv'
+
+require 'fastnn'
+require 'libhtkio'
+require 'threads'
+
+dofile("fastnn/fastnn_baseline.lua")
+
+env = string.format([[
+package.path="/home/slhome/wd007/.luarocks/share/lua/5.1/?.lua;/home/slhome/wd007/.luarocks/share/lua/5.1/?/init.lua;/sgfs/users/wd007/src/nerv/install/share/lua/5.1/?.lua;/sgfs/users/wd007/src/nerv/install/share/lua/5.1/?/init.lua;"..package.path;
+package.cpath="/home/slhome/wd007/.luarocks/lib/lua/5.1/?.so;/sgfs/users/wd007/src/nerv/install/lib/lua/5.1/?.so;"..package.cpath
+local k,l,_=pcall(require,"luarocks.loader") _=k and l.add_context("nerv","scm-1")
+]])
+
+
+train_thread_code = [[
+%s
+
+require 'nerv'
+require 'fastnn'
+dofile("fastnn/fastnn_baseline.lua")
+os.execute("export MALLOC_CHECK_=0")
+
+local thread_idx = %d
+local feat_repo_shareid = %d
+local data_mutex_shareid = %d
+local master_shareid = %d
+local gpu_shareid = %d
+local xent_shareid = %d
+local batch_size = %d
+local lrate = %f
+local bp = %d
+local scp_file = '%s'
+local nnet_in = '%s'
+local nnet_out = '%s'
+
+local share_mutex = threads.Mutex(data_mutex_shareid)
+local share_master = fastnn.ModelSync(master_shareid)
+local share_gpu = fastnn.CDevice(gpu_shareid)
+local share_xent = fastnn.CXent(xent_shareid)
+
+if bp == 0 then
+ bp = false
+else
+ bp = true
+ gconf.tr_scp = scp_file
+end
+
+gconf.randomize = bp
+gconf.lrate = lrate
+gconf.batch_size = batch_size
+gconf.network[1] = nnet_in
+nerv.info_stderr("input network: %%s", gconf.network[1])
+--nerv.info_stderr(gconf.randomize)
+nerv.info_stderr("input batch_size: %%d", gconf.batch_size)
+nerv.info_stderr("input scp_file: %%s", scp_file)
+nerv.info_stderr("input lrate: %%f", gconf.lrate)
+
+
+share_mutex:lock()
+share_gpu:select_gpu()
+
+nerv.context = nerv.CCuContext()
+--print(nerv.context)
+
+nerv.info_stderr("thread %%d loading transf ...", thread_idx)
+local param_transf_repo = nerv.ParamRepo()
+param_transf_repo:import(gconf.transf, nil, gconf)
+local transf_node_repo = make_transf_node_repo(param_transf_repo)
+local transf_layer_repo = make_transf_link_repo(transf_node_repo, param_transf_repo)
+local transf = transf_layer_repo:get_layer("global_transf")
+
+nerv.info_stderr("thread %%d loading network ...", thread_idx)
+local param_network_repo = nerv.ParamRepo()
+param_network_repo:import(gconf.network, nil, gconf)
+local network_node_repo = make_network_node_repo(param_network_repo)
+local network_layer_repo = make_network_link_repo(network_node_repo, param_network_repo)
+local network = get_network(network_layer_repo)
+share_mutex:unlock()
+
+local buffer = make_buffer(make_readers(nil, transf_layer_repo, feat_repo_shareid, data_mutex_shareid))
+
+local input_order = get_input_order()
+
+ -- initialize the network
+ network:init(gconf.batch_size)
+ gconf.cnt = 0
+ err_input = {nerv.CuMatrixFloat(gconf.batch_size, 1)}
+ err_input[1]:fill(1)
+
+ share_master:Initialize(network)
+ share_master:SyncInc()
+
+ for data in buffer.get_data, buffer do
+
+ gconf.cnt = gconf.cnt + 1
+ if gconf.cnt == 2000 then
+ print_stat(network_node_repo)
+ gconf.cnt = 0
+ end
+
+ local input = {}
+
+ for i, id in ipairs(input_order) do
+ if data[id] == nil then
+ nerv.error("input data %%s not found", id)
+ end
+ table.insert(input, data[id])
+ end
+
+ local output = {nerv.CuMatrixFloat(gconf.batch_size, 1)}
+ err_output = {input[1]:create()}
+ network:propagate(input, output)
+
+ if bp then
+ network:back_propagate(err_input, err_output, input, output)
+ network:gradient(err_input, input, output)
+
+ share_master:LockModel()
+ share_master:WeightToD(network)
+ network:update_gradient()
+ -- network:update(err_input, input, output)
+ share_master:WeightFromD(network)
+ share_master:UnLockModel()
+ end
+
+ -- collect garbage in-time to save GPU memory
+ collectgarbage("collect")
+ end
+
+ --print_stat(network_node_repo)
+ local ce_crit = network_node_repo:get_layer("ce_crit")
+ local xent = fastnn.CXent(ce_crit.total_frames, ce_crit.total_correct, ce_crit.total_ce, ce_crit.total_ce)
+
+ share_master:LockModel()
+ share_xent:add(xent)
+ share_master:SyncDec()
+ --print(string.format("ThreadCount: %%d", share_master:ThreadCount()))
+ if share_master:ThreadCount() == 0 and bp then
+ share_master:WeightToD(network)
+ local fname = string.format("%%s_tr%%.3f",
+ nnet_out, frame_acc(share_xent))
+ nerv.info_stderr("writing back %%s ...", fname)
+ network:get_params():export(fname, nil)
+ end
+ share_master:UnLockModel()
+]]
+
+
+function get_train_thread(train_thread_code, env, thread_idx, feat_repo_shareid,
+ data_mutex_shareid, master_shareid, gpu_shareid, xent_shareid,
+ batch_size, lrate, bp, scp_file, nnet_in, nnet_out)
+ return string.format(train_thread_code, env, thread_idx, feat_repo_shareid,
+ data_mutex_shareid, master_shareid, gpu_shareid, xent_shareid,
+ batch_size, lrate, bp, scp_file, nnet_in, nnet_out)
+end
+
+function trainer(batch_size, lrate, bp, scp_file, nnet_in, nnet_out, num_threads)
+ local train_threads={}
+ local trainer = {}
+ local num_threads=num_threads
+
+ local data_mutex = threads.Mutex()
+ local data_mutex_shareid = data_mutex:id()
+
+ local master = fastnn.CModelSync()
+ local master_shareid = master:id()
+ --print(master)
+
+ local xent = fastnn.CXent()
+ local xent_shareid = xent:id()
+ --print(xent)
+
+ local gpu = fastnn.CDevice()
+ local gpu_shareid = gpu:id()
+ --print(gpu_shareid)
+ gpu:init()
+
+ local feat_repo = nerv.TNetFeatureRepo(scp_file, gconf.htk_conf, gconf.frm_ext)
+ local feat_repo_shareid = feat_repo:id()
+
+ for i=1,num_threads,1 do
+
+ train_threads[i] = get_train_thread(train_thread_code, env, i, feat_repo_shareid,
+ data_mutex_shareid, master_shareid, gpu_shareid, xent_shareid,
+ batch_size, lrate, bp, scp_file, nnet_in, nnet_out)
+ --print(train_threads[i])
+ trainer[i] = threads.Thread(train_threads[i])
+ end
+
+ nerv.info_stderr('| waiting for thread...')
+
+ for i=1,num_threads,1 do
+ trainer[i]:free()
+ end
+
+ print_xent(xent)
+
+ nerv.info_stderr('| all thread finished!')
+
+ return frame_acc(xent)
+end
+
+function get_filename(fname)
+ return string.gsub((string.gsub(fname, "(.*/)(.*)", "%2")),"(.*)%..*", "%1")
+end
+
+function do_sds(tr_scp, sds_scp, sds_rate)
+ math.randomseed(os.time())
+ local scp_file = io.open(tr_scp, "r")
+ local sds_file = io.open(sds_scp, "w")
+ for line in scp_file:lines() do
+ rate = math.random()
+ if (rate < sds_rate) then
+ sds_file:write(line.."\n")
+ end
+ end
+ scp_file:close()
+ sds_file:close()
+end
+
+function print_tag(iter)
+ io.stderr:write(string.format("########################################################\n"))
+ io.stderr:write(string.format("# NN TRAINING ITERATION:%d, %s\n", iter, os.date()))
+ io.stderr:write(string.format("########################################################\n"))
+end
+
+
+start_halving_inc = 0.5
+halving_factor = 0.8
+end_halving_inc = 0.1
+min_iter = 1
+max_iter = 20
+min_halving = 0
+gconf.batch_size = 256
+pf0 = get_filename(gconf.network[1])
+nnet_in = gconf.network[1]
+nnet_out = ""
+sds_scp = "tr_sds_"..string.format("%.4d", math.random()*10000)..".scp" --"tr_sds.scp"
+sds_factor = 0.4
+num_threads = 2
+global_option = nil
+
+os.execute("export MALLOC_CHECK_=0")
+print_gconf()
+
+-- training begin
+nerv.info_stderr("begin initial cross validation")
+accu_best = trainer(gconf.batch_size, gconf.lrate, 0,
+ gconf.cv_scp, nnet_in, "", num_threads)
+local do_halving = false
+local accu_new = accu_best
+
+nerv.info_stderr("initial cross validation: %.3f\n", accu_best)
+
+for i = 1, max_iter do
+
+ if accu_new >= accu_best then
+ local sds_rate = math.cos((i-1)*11.0/180*math.pi)
+ if (sds_rate <= sds_factor) then
+ sds_rate = sds_factor
+ end
+ nerv.info_stderr("iteration %d sds_rate: %.6f", i, sds_rate)
+ do_sds(gconf.tr_scp, sds_scp, sds_rate)
+ end
+
+ nnet_out=pf0.."_iter"..i
+ --print(nnet_out)
+ print_tag(i)
+ nerv.info_stderr("[NN] begin iteration %d learning_rate: %.3f batch_size: %d.", i, gconf.lrate, gconf.batch_size)
+ accu_tr = trainer(gconf.batch_size, gconf.lrate, 1,
+ sds_scp, nnet_in, nnet_out, num_threads)
+ collectgarbage("collect")
+ nerv.info_stderr("[TR] end iteration %d frame_accuracy: %.3f.\n", i, accu_tr)
+ os.execute("sleep " .. 3)
+
+ nnet_out = nnet_out.."_tr"..accu_tr
+ accu_new = trainer(gconf.batch_size, gconf.lrate, 0,
+ gconf.cv_scp, nnet_out, "", num_threads)
+ collectgarbage("collect")
+ nerv.info_stderr("[CV] end iteration %d frame_accuracy: %.3f.\n\n", i, accu_new)
+ os.execute("sleep " .. 3)
+
+ local nnet_tmp = string.format("%s_%s_iter_%d_lr%f_tr%.3f_cv%.3f",
+ pf0,
+ os.date("%Y%m%d%H%M%S"),
+ i, gconf.lrate, accu_tr, accu_new)
+
+ -- TODO: revert the weights
+ local accu_diff = accu_new - accu_best
+ local cmd
+ if accu_new > accu_best then
+ accu_best = accu_new
+ nnet_in = nnet_tmp
+ gconf.batch_size = gconf.batch_size + 128
+ if gconf.batch_size > 1024 then
+ gconf.batch_size = 1024
+ end
+ else
+ -- reject
+ nnet_tmp = nnet_tmp.."_rejected"
+ do_halving = true
+ end
+ cmd = "mv "..nnet_out.." "..nnet_tmp
+ os.execute(cmd)
+
+ if do_halving and accu_diff < end_halving_inc and i > min_iter then
+ break;
+ end
+
+ if accu_diff < start_halving_inc and i >= min_halving then
+ do_halving = true
+ end
+
+ if do_halving then
+ gconf.lrate = gconf.lrate * halving_factor
+ halving_factor = halving_factor - 0.025
+ if halving_factor < 0.6 then
+ halving_factor = 0.6
+ end
+ end
+ nerv.info_stderr("iteration %d done!", i)
+end
+
+
diff --git a/fastnn/fastnn-scm-1.rockspec b/fastnn/fastnn-scm-1.rockspec
new file mode 100644
index 0000000..69dfb60
--- /dev/null
+++ b/fastnn/fastnn-scm-1.rockspec
@@ -0,0 +1,36 @@
+package = "fastnn"
+version = "scm-1"
+source = {
+ url = "https://github.com/uphantom/nerv-fastnn.git"
+}
+description = {
+ summary = "speed up Nerv neural network training.",
+ detailed = [[
+ ]],
+ homepage = "https://github.com/uphantom/nerv-fastnn",
+ license = "BSD"
+}
+dependencies = {
+ "nerv >= scm-1",
+ "lua >= 5.1"
+}
+build = {
+ type = "make",
+ build_variables = {
+ CFLAGS="$(CFLAGS)",
+ LIBFLAG="$(LIBFLAG)",
+ LUA_LIBDIR="$(LUA_LIBDIR)",
+ LUA_BINDIR="$(LUA_BINDIR)",
+ LUA_INCDIR="$(LUA_INCDIR)",
+ INST_PREFIX="$(PREFIX)",
+ LUA="$(LUA)",
+ },
+ install_variables = {
+ LUA_BINDIR="$(LUA_BINDIR)",
+ INST_PREFIX="$(PREFIX)",
+ INST_BINDIR="$(BINDIR)",
+ INST_LIBDIR="$(LIBDIR)",
+ INST_LUADIR="$(LUADIR)",
+ INST_CONFDIR="$(CONFDIR)",
+ },
+}
diff --git a/fastnn/fastnn_baseline.lua b/fastnn/fastnn_baseline.lua
new file mode 100644
index 0000000..80b1d6b
--- /dev/null
+++ b/fastnn/fastnn_baseline.lua
@@ -0,0 +1,236 @@
+require 'htk_io'
+
+gconf = {lrate = 0.2, wcost = 1e-6, momentum = 0.9,
+ cumat_type = nerv.CuMatrixFloat,
+ mmat_type = nerv.MMatrixFloat,
+ frm_ext = 5,
+ batch_size = 256,
+ buffer_size = 81920,
+ tr_scp = "/sgfs/users/wd007/asr/baseline_chn_50h/finetune/finetune_baseline/train.scp",
+ cv_scp = "/sgfs/users/wd007/asr/baseline_chn_50h/finetune/finetune_baseline/train_cv.scp",
+ htk_conf = "/sgfs/users/wd007/asr/baseline_chn_50h/finetune/finetune_baseline/fbank_d_a_z.conf",
+ transf = {"/sgfs/users/wd007/src/nerv/tools/nerv.global.transf"},
+ network = {"/sgfs/users/wd007/src/nerv/tools/nerv.svd0.55_3000h_iter1.init"},
+ debug = false}
+
+function make_transf_node_repo(param_transf_repo)
+ return nerv.LayerRepo(
+ {
+ -- global transf
+ ["nerv.BiasLayer"] =
+ {
+ blayer1 = {{bias = "bias1"}, {dim_in = {1320}, dim_out = {1320}}},
+ },
+ ["nerv.WindowLayer"] =
+ {
+ wlayer1 = {{window = "window1"}, {dim_in = {1320}, dim_out = {1320}}},
+ },
+
+ }, param_transf_repo, gconf)
+end
+
+function make_network_node_repo(param_network_repo)
+ return nerv.LayerRepo(
+ {
+ -- biased linearity
+ ["nerv.AffineLayer"] =
+ {
+ affine0 = {{ltp = "affine0_ltp", bp = "affine0_bp"},
+ {dim_in = {1320}, dim_out = {2048}}},
+ affine1 = {{ltp = "affine1_ltp", bp = "affine1_bp"},
+ {dim_in = {2048}, dim_out = {367}}},
+ affine2 = {{ltp = "affine2_ltp", bp = "affine2_bp"},
+ {dim_in = {367}, dim_out = {2048}}},
+ affine3 = {{ltp = "affine3_ltp", bp = "affine3_bp"},
+ {dim_in = {2048}, dim_out = {408}}},
+ affine4 = {{ltp = "affine4_ltp", bp = "affine4_bp"},
+ {dim_in = {408}, dim_out = {2048}}},
+ affine5 = {{ltp = "affine5_ltp", bp = "affine5_bp"},
+ {dim_in = {2048}, dim_out = {368}}},
+ affine6 = {{ltp = "affine6_ltp", bp = "affine6_bp"},
+ {dim_in = {368}, dim_out = {2048}}},
+ affine7 = {{ltp = "affine7_ltp", bp = "affine7_bp"},
+ {dim_in = {2048}, dim_out = {303}}},
+ affine8 = {{ltp = "affine8_ltp", bp = "affine8_bp"},
+ {dim_in = {303}, dim_out = {2048}}},
+ affine9 = {{ltp = "affine9_ltp", bp = "affine9_bp"},
+ {dim_in = {2048}, dim_out = {277}}},
+ affine10 = {{ltp = "affine10_ltp", bp = "affine10_bp"},
+ {dim_in = {277}, dim_out = {2048}}},
+ affine11 = {{ltp = "affine11_ltp", bp = "affine11_bp"},
+ {dim_in = {2048}, dim_out = {361}}},
+ affine12 = {{ltp = "affine12_ltp", bp = "affine12_bp"},
+ {dim_in = {361}, dim_out = {2048}}},
+ affine13 = {{ltp = "affine13_ltp", bp = "affine13_bp"},
+ {dim_in = {2048}, dim_out = {441}}},
+ affine14 = {{ltp = "affine14_ltp", bp = "affine14_bp"},
+ {dim_in = {441}, dim_out = {10092}}},
+ },
+ ["nerv.SigmoidLayer"] =
+ {
+ sigmoid0 = {{}, {dim_in = {2048}, dim_out = {2048}}},
+ sigmoid1 = {{}, {dim_in = {2048}, dim_out = {2048}}},
+ sigmoid2 = {{}, {dim_in = {2048}, dim_out = {2048}}},
+ sigmoid3 = {{}, {dim_in = {2048}, dim_out = {2048}}},
+ sigmoid4 = {{}, {dim_in = {2048}, dim_out = {2048}}},
+ sigmoid5 = {{}, {dim_in = {2048}, dim_out = {2048}}},
+ sigmoid6 = {{}, {dim_in = {2048}, dim_out = {2048}}},
+ },
+ ["nerv.SoftmaxCELayer"] =
+ {
+ ce_crit = {{}, {dim_in = {10092, 1}, dim_out = {1}, compressed = true}}
+ }
+
+ }, param_network_repo, gconf)
+end
+
+
+function make_transf_link_repo(layer_transf_node_repo, param_transf_repo)
+ return nerv.LayerRepo(
+ {
+ ["nerv.DAGLayer"] =
+ {
+ global_transf = {{}, {
+ dim_in = {1320}, dim_out = {1320},
+ sub_layers = layer_transf_node_repo,
+ connections =
+ {
+ ["<input>[1]"] = "blayer1[1]",
+ ["blayer1[1]"] = "wlayer1[1]",
+ ["wlayer1[1]"] = "<output>[1]"
+ }
+ }},
+
+ }
+ }, param_transf_repo, gconf)
+end
+
+
+function make_network_link_repo(layer_network_node_repo, param_network_repo)
+ return nerv.LayerRepo(
+ {
+ ["nerv.DAGLayer"] =
+ {
+ main = {{}, {
+ dim_in = {1320, 1}, dim_out = {1},
+ sub_layers = layer_network_node_repo,
+ connections = {
+ ["<input>[1]"] = "affine0[1]",
+ ["affine0[1]"] = "sigmoid0[1]",
+ ["sigmoid0[1]"] = "affine1[1]",
+ ["affine1[1]"] = "affine2[1]",
+ ["affine2[1]"] = "sigmoid1[1]",
+ ["sigmoid1[1]"] = "affine3[1]",
+ ["affine3[1]"] = "affine4[1]",
+ ["affine4[1]"] = "sigmoid2[1]",
+ ["sigmoid2[1]"] = "affine5[1]",
+ ["affine5[1]"] = "affine6[1]",
+ ["affine6[1]"] = "sigmoid3[1]",
+ ["sigmoid3[1]"] = "affine7[1]",
+ ["affine7[1]"] = "affine8[1]",
+ ["affine8[1]"] = "sigmoid4[1]",
+ ["sigmoid4[1]"] = "affine9[1]",
+ ["affine9[1]"] = "affine10[1]",
+ ["affine10[1]"] = "sigmoid5[1]",
+ ["sigmoid5[1]"] = "affine11[1]",
+ ["affine11[1]"] = "affine12[1]",
+ ["affine12[1]"] = "sigmoid6[1]",
+ ["sigmoid6[1]"] = "affine13[1]",
+ ["affine13[1]"] = "affine14[1]",
+ ["affine14[1]"] = "ce_crit[1]",
+ ["<input>[2]"] = "ce_crit[2]",
+ ["ce_crit[1]"] = "<output>[1]"
+ }
+ }}
+ }
+ }, param_network_repo, gconf)
+end
+
+function get_network(layer_repo)
+ return layer_repo:get_layer("main")
+end
+
+
+function make_readers(scp_file, layer_repo, feat_repo_shareid, data_mutex_shareid)
+ return {
+ {reader = nerv.TNetReader(gconf,
+ {
+ id = "main_scp",
+ scp_file = scp_file,
+ conf_file = gconf.htk_conf,
+ frm_ext = gconf.frm_ext,
+ mlfs = {
+ phone_state = {
+ file = "/sgfs/users/wd007/asr/baseline_chn_50h/finetune/finetune_baseline/ref.mlf",
+ format = "map",
+ format_arg = "/sgfs/users/wd007/asr/baseline_chn_50h/finetune/finetune_baseline/dict",
+ dir = "*/",
+ ext = "lab"
+ }
+ },
+ global_transf = layer_repo:get_layer("global_transf")
+ }, feat_repo_shareid, data_mutex_shareid),
+ data = {main_scp = 1320, phone_state = 1}}
+ }
+end
+
+
+function make_buffer(readers)
+ return nerv.SGDBuffer(gconf,
+ {
+ buffer_size = gconf.buffer_size,
+ randomize = gconf.randomize,
+ readers = readers
+ })
+end
+
+function get_feat_id()
+ return {main_scp = true}
+end
+
+function get_input_order()
+ return {"main_scp", "phone_state"}
+end
+
+function get_accuracy(sublayer_repo)
+ local ce_crit = sublayer_repo:get_layer("ce_crit")
+ return ce_crit.total_correct / ce_crit.total_frames * 100
+end
+
+function print_stat(sublayer_repo)
+ local ce_crit = sublayer_repo:get_layer("ce_crit")
+ nerv.info("*** training stat begin ***")
+ nerv.printf("cross entropy:\t\t%.8f\n", ce_crit.total_ce)
+ nerv.printf("correct:\t\t%d\n", ce_crit.total_correct)
+ nerv.printf("frames:\t\t\t%d\n", ce_crit.total_frames)
+ nerv.printf("err/frm:\t\t%.8f\n", ce_crit.total_ce / ce_crit.total_frames)
+ nerv.printf("accuracy:\t\t%.3f%%\n", get_accuracy(sublayer_repo))
+ nerv.info("*** training stat end ***")
+end
+
+function print_xent(xent)
+ local totalframes = xent:totalframes()
+ local loss = xent:loss()
+ local correct = xent:correct()
+ nerv.info_stderr("*** training statistics info begin ***")
+ nerv.info_stderr("total frames:\t\t%d", totalframes)
+ nerv.info_stderr("cross entropy:\t%.8f", loss/totalframes)
+ nerv.info_stderr("frame accuracy:\t%.3f%%", 100*correct/totalframes)
+ nerv.info_stderr("*** training statistics info end ***")
+end
+
+function frame_acc(xent)
+ local correct = xent:correct()
+ local totalframes = xent:totalframes()
+ return string.format("%.3f", 100*correct/totalframes)
+end
+
+function print_gconf()
+ nerv.info_stderr("%s \t:= %s", "network", gconf.network[1])
+ nerv.info_stderr("%s \t:= %s", "transf", gconf.transf[1])
+ nerv.info_stderr("%s \t:= %s", "batch_size", gconf.batch_size)
+ nerv.info_stderr("%s \t:= %s", "buffer_size", gconf.buffer_size)
+ nerv.info_stderr("%s \t:= %s", "lrate", gconf.lrate)
+ nerv.info_stderr("%s \t:= %s", "tr_scp", gconf.tr_scp)
+ nerv.info_stderr("%s \t:= %s", "cv_scp", gconf.cv_scp)
+end
diff --git a/fastnn/init.c b/fastnn/init.c
new file mode 100644
index 0000000..2022293
--- /dev/null
+++ b/fastnn/init.c
@@ -0,0 +1,43 @@
+#include <lua.h>
+#include <lauxlib.h>
+
+#if LUA_VERSION_NUM == 501
+static void luaL_setfuncs(lua_State *L, const luaL_Reg *l, int nup)
+{
+ luaL_checkstack(L, nup+1, "too many upvalues");
+ for (; l->name != NULL; l++) { /* fill the table with given functions */
+ int i;
+ lua_pushstring(L, l->name);
+ for (i = 0; i < nup; i++) /* copy upvalues to the top */
+ lua_pushvalue(L, -(nup+1));
+ lua_pushcclosure(L, l->func, nup); /* closure with those upvalues */
+ lua_settable(L, -(nup + 3));
+ }
+ lua_pop(L, nup); /* remove upvalues */
+}
+#endif
+
+
+extern void fastnn_init_modelsync(lua_State *L);
+extern void fastnn_init_example(lua_State *L);
+extern void fastnn_init_example_repo(lua_State *L);
+extern void fastnn_init_device(lua_State *L);
+extern void fastnn_init_context(lua_State *L);
+
+int luaopen_libfastnn(lua_State *L)
+{
+ //printf("%s %lx\n", model_sync__[13].name, model_sync__[13].func);
+ lua_newtable(L);
+ /* duplicate table */
+ lua_pushvalue(L, -1);
+ /* set table to global index */
+ lua_setfield(L, LUA_GLOBALSINDEX, "fastnn");
+
+ fastnn_init_modelsync(L);
+ fastnn_init_example(L);
+ fastnn_init_example_repo(L);
+ fastnn_init_device(L);
+ fastnn_init_context(L);
+ return 1;
+}
+
diff --git a/fastnn/init.lua b/fastnn/init.lua
new file mode 100644
index 0000000..5e290ca
--- /dev/null
+++ b/fastnn/init.lua
@@ -0,0 +1,11 @@
+require 'libfastnn'
+
+function fastnn.include(filename)
+ local caller = debug.getinfo(2, "S").source:sub(2)
+ dofile(nerv.dirname(caller) .. filename)
+end
+
+fastnn.include('lib/modelsync.lua')
+fastnn.include('io/example.lua')
+fastnn.include('device/device.lua')
+
diff --git a/fastnn/io/Example.cpp b/fastnn/io/Example.cpp
new file mode 100644
index 0000000..8f200b7
--- /dev/null
+++ b/fastnn/io/Example.cpp
@@ -0,0 +1,186 @@
+
+#include <deque>
+#include <vector>
+#include <string>
+
+
+extern "C" {
+
+#include "../threads/lib/THThread.h"
+#include "Example.h"
+#include "stdlib.h"
+#include "stdio.h"
+
+#include "../../nerv/lib/matrix/generic/elem_type.h"
+#include "common.h"
+
+extern Matrix* nerv_matrix_cuda_float_create(long nrow, long ncol, Status *status);
+void nerv_matrix_cuda_float_copy_fromd(Matrix *a, const Matrix *b,
+ int a_begin, int b_begin, int b_end,
+ Status *status);
+
+struct Example
+{
+ std::vector<Matrix *> inputs;
+ std::string id;
+ int refcount;
+};
+
+struct ExamplesRepository
+{
+ int buffer_size_;
+ THSemaphore *full_semaphore_;
+ THSemaphore *empty_semaphore_;
+ THMutex *examples_mutex_;
+
+ std::deque<Example*> examples_;
+ bool done_;
+ int refcount;
+ int gpuid;
+};
+
+Example* Example_new()
+{
+ Example *example = new Example; //(Example*)malloc(sizeof(Example));
+ example->refcount = 1;
+ return example;
+}
+
+Example* Example_newWithId(long id)
+{
+ Example* example = (Example*)(id);
+ __sync_fetch_and_add(&example->refcount, 1);
+ return example;
+}
+
+long Example_id(Example *example)
+{
+ return (long)(example);
+}
+
+void Example_destroy(Example* example)
+{
+ //printf("Example_destroy: %d\n", example->inputs.size());
+ if (NULL != example && __sync_fetch_and_add(&example->refcount, -1) == 1)
+ {
+ delete example;
+ example = NULL;
+ }
+}
+
+int Example_size(Example* example)
+{
+ return example->inputs.size();
+}
+
+Matrix* Example_at(Example* example, int idx)
+{
+ return example->inputs.at(idx);
+}
+
+void Example_pushback(Example* example, Matrix* m)
+{
+ Status status;
+ Matrix *newm = nerv_matrix_cuda_float_create(m->nrow, m->ncol, &status);
+ nerv_matrix_cuda_float_copy_fromd(newm, m, 0, 0, m->nrow, &status);
+ //__sync_fetch_and_add(&m->refcount, 1);
+ return example->inputs.push_back(newm);
+}
+
+//////////////////////////////////////////////////////////////
+
+
+ExamplesRepository* ExamplesRepository_new(int buffersize)
+{
+ ExamplesRepository *repo = new ExamplesRepository; //(ExamplesRepository*)malloc(sizeof(ExamplesRepository));
+ repo->buffer_size_ = buffersize;
+ repo->full_semaphore_ = THSemaphore_new(0);
+ repo->empty_semaphore_ = THSemaphore_new(buffersize);
+ repo->examples_mutex_ = THMutex_new();
+// repo->examples_ = new std::deque<Example*>();
+ repo->done_ = false;
+ repo->refcount = 1;
+ repo->gpuid = -1;
+ return repo;
+}
+
+ExamplesRepository* ExamplesRepository_newWithId(long id)
+{
+ ExamplesRepository *repo = (ExamplesRepository*)(id);
+ __sync_fetch_and_add(&repo->refcount, 1);
+ return repo;
+}
+
+long ExamplesRepository_id(ExamplesRepository* repo)
+{
+ return (long)(repo);
+}
+
+int ExamplesRepository_getGpuId(ExamplesRepository* repo)
+{
+ return repo->gpuid;
+}
+
+void ExamplesRepository_setGpuId(ExamplesRepository* repo, int gpuid)
+{
+ repo->gpuid = gpuid;
+}
+
+void ExamplesRepository_destroy(ExamplesRepository* repo)
+{
+ if (NULL != repo && __sync_fetch_and_add(&repo->refcount, -1) == 1)
+ {
+ if (repo->full_semaphore_)
+ THSemaphore_free(repo->full_semaphore_);
+ if (repo->empty_semaphore_)
+ THSemaphore_free(repo->empty_semaphore_);
+ if (repo->examples_mutex_)
+ THMutex_free(repo->examples_mutex_);
+ delete repo;
+ repo = NULL;
+ }
+}
+
+void AcceptExample(ExamplesRepository *repo, Example *example)
+{
+ THSemaphore_wait(repo->empty_semaphore_);
+ THMutex_lock(repo->examples_mutex_);
+ __sync_fetch_and_add(&example->refcount, 1);
+ repo->examples_.push_back(example);
+ THMutex_unlock(repo->examples_mutex_);
+ THSemaphore_signal(repo->full_semaphore_);
+}
+
+void ExamplesDone(ExamplesRepository *repo)
+{
+ for (int i = 0; i < repo->buffer_size_; i++)
+ THSemaphore_wait(repo->empty_semaphore_);
+
+ repo->done_ = true;
+ THSemaphore_signal(repo->full_semaphore_);
+}
+
+Example* ProvideExample(ExamplesRepository *repo)
+{
+ Example *ans = NULL;
+ THSemaphore_wait(repo->full_semaphore_);
+ if (repo->done_)
+ {
+ THSemaphore_signal(repo->full_semaphore_); // Increment the semaphore so
+ // the call by the next thread will not block.
+ return NULL; // no examples to return-- all finished.
+ }
+ else
+ {
+ THMutex_lock(repo->examples_mutex_);
+ ans = repo->examples_.front();
+ repo->examples_.pop_front();
+ THMutex_unlock(repo->examples_mutex_);
+ THSemaphore_signal(repo->empty_semaphore_);
+ }
+ return ans;
+}
+
+}
+
+
diff --git a/fastnn/io/Example.h b/fastnn/io/Example.h
new file mode 100644
index 0000000..1c462ab
--- /dev/null
+++ b/fastnn/io/Example.h
@@ -0,0 +1,49 @@
+#ifndef NERV_FASTNN_EXAMPLE_H
+#define NERV_FASTNN_EXAMPLE_H
+
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "matrix/matrix.h"
+#include "stdbool.h"
+#define STRLEN 1024
+
+
+
+typedef struct ExamplesRepository ExamplesRepository;
+typedef struct Example Example;
+
+ExamplesRepository* ExamplesRepository_new(int buffersize);
+ExamplesRepository* ExamplesRepository_newWithId(long id);
+long ExamplesRepository_id(ExamplesRepository* reop);
+void ExamplesRepository_destroy(ExamplesRepository* reop);
+void AcceptExample(ExamplesRepository *repo, Example *example);
+void ExamplesDone(ExamplesRepository *repo);
+Example* ProvideExample(ExamplesRepository *repo);
+void ExamplesRepository_setGpuId(ExamplesRepository* repo, int gpuid);
+int ExamplesRepository_getGpuId(ExamplesRepository* repo);
+
+
+Example* Example_new();
+Example* Example_newWithId(long id);
+long Example_id(Example* example);
+void Example_destroy(Example* example);
+int Example_size(Example* example);
+Matrix* Example_at(Example* example, int idx);
+void Example_pushback(Example* example, Matrix* m);
+
+
+
+
+
+
+
+
+#ifdef __cplusplus
+} // closing brace for extern "C"
+
+#endif // end Example
+
+#endif
diff --git a/fastnn/io/example.c b/fastnn/io/example.c
new file mode 100644
index 0000000..f2f0916
--- /dev/null
+++ b/fastnn/io/example.c
@@ -0,0 +1,249 @@
+
+#include <stdio.h>
+#include <stdlib.h>
+
+#include <lua.h>
+#include <lualib.h>
+#include <luaT/luaT.h>
+
+
+
+#include "../threads/lib/luaTHRD.h"
+#include "Example.h"
+
+
+const char *fastnn_example_repo_tname = "fastnn.CExamplesRepo";
+const char *fastnn_example_tname = "fastnn.CExample";
+
+static int example_repo_new(lua_State *L)
+{
+ ExamplesRepository *repo = NULL ;
+ int buffersize;
+ bool shared;
+ long id;
+ if(lua_gettop(L) == 2 && lua_isboolean(L, 2))
+ {
+ shared = lua_toboolean(L, 2);
+ if (shared)
+ {
+ id = luaL_checkinteger(L, 1);
+ repo = ExamplesRepository_newWithId(id);
+ }
+ else
+ {
+ buffersize = luaL_checkinteger(L, 1);
+ repo = ExamplesRepository_new(buffersize);
+ }
+ }
+ else
+ luaL_error(L, "example: example repo new invalid arguments");
+ if (!repo)
+ luaL_error(L, "example: example new failed");
+
+ luaTHRD_pushudata(L, repo, fastnn_example_repo_tname);
+
+ return 1;
+}
+
+static int example_repo_newfromid(lua_State *L)
+{
+ ExamplesRepository *repo = NULL ;
+ if(lua_gettop(L) == 1)
+ {
+ long id = luaL_checkinteger(L, 1);
+ repo = ExamplesRepository_newWithId(id);
+ }
+ else
+ luaL_error(L, "example: example repo new invalid arguments");
+ if (!repo)
+ luaL_error(L, "example: example new failed");
+
+ luaTHRD_pushudata(L, repo, fastnn_example_repo_tname);
+
+ return 1;
+}
+
+static int example_repo_tostring(lua_State *L)
+{
+ char str[STRLEN];
+ ExamplesRepository *repo = luaTHRD_checkudata(L, 1, fastnn_example_repo_tname);
+ snprintf(str, STRLEN, "%s <%lx>", fastnn_example_repo_tname, ExamplesRepository_id(repo));
+ lua_pushstring(L, str);
+ return 1;
+}
+
+static int example_repo_id(lua_State *L)
+{
+ ExamplesRepository *repo = luaTHRD_checkudata(L, 1, fastnn_example_repo_tname);
+ lua_pushinteger(L, ExamplesRepository_id(repo));
+ return 1;
+}
+
+static int example_repo_get_gpuid(lua_State *L)
+{
+ ExamplesRepository *repo = luaTHRD_checkudata(L, 1, fastnn_example_repo_tname);
+ lua_pushinteger(L, ExamplesRepository_getGpuId(repo));
+ return 1;
+}
+
+static int example_repo_set_gpuid(lua_State *L)
+{
+ ExamplesRepository *repo = luaTHRD_checkudata(L, 1, fastnn_example_repo_tname);
+ int gpuid = luaL_checkinteger(L, 2);
+ ExamplesRepository_setGpuId(repo, gpuid);
+ return 0;
+}
+
+static int example_repo_accept(lua_State *L)
+{
+ ExamplesRepository *repo = luaTHRD_checkudata(L, 1, fastnn_example_repo_tname);
+ Example *example = luaTHRD_checkudata(L, 2, fastnn_example_tname);
+ AcceptExample(repo, example);
+ return 0;
+}
+
+static int example_repo_provide(lua_State *L)
+{
+ ExamplesRepository *repo = luaTHRD_checkudata(L, 1, fastnn_example_repo_tname);
+ Example *example = ProvideExample(repo);
+ if (NULL != example)
+ {
+ luaTHRD_pushudata(L, example, fastnn_example_tname);
+ return 1;
+ }
+ else
+ return 0;
+}
+
+static int example_repo_done(lua_State *L)
+{
+ ExamplesRepository *repo = luaTHRD_checkudata(L, 1, fastnn_example_repo_tname);
+ ExamplesDone(repo);
+ return 0;
+}
+
+static int example_repo_destroy(lua_State *L)
+{
+ ExamplesRepository *repo = luaTHRD_checkudata(L, 1, fastnn_example_repo_tname);
+ ExamplesRepository_destroy(repo);
+ return 0;
+}
+
+/******************************************/
+
+static int example_new(lua_State *L)
+{
+
+ Example *example = NULL;
+ if(lua_gettop(L) == 0)
+ {
+ example = Example_new();
+ }
+ else if(lua_gettop(L) == 1)
+ {
+ long id = luaL_checkinteger(L, 1);
+ example = Example_newWithId(id);
+ }
+ else
+ luaL_error(L, "example: example new invalid arguments");
+ if (!example)
+ luaL_error(L, "example: example failed");
+
+ luaTHRD_pushudata(L, example, fastnn_example_tname);
+
+ return 1;
+}
+
+static int example_id(lua_State *L)
+{
+ Example *example = luaTHRD_checkudata(L, 1, fastnn_example_tname);
+ lua_pushinteger(L, Example_id(example));
+ return 1;
+}
+
+static int example_tostring(lua_State *L)
+{
+ char str[STRLEN];
+ Example* example = luaTHRD_checkudata(L, 1, fastnn_example_tname);
+ snprintf(str, STRLEN, "%s <%lx>", fastnn_example_tname, Example_id(example));
+ lua_pushstring(L, str);
+ return 1;
+}
+
+static int example_size(lua_State *L)
+{
+ Example *example = luaTHRD_checkudata(L, 1, fastnn_example_tname);
+ lua_pushinteger(L, Example_size(example));
+ return 1;
+}
+
+static int example_at(lua_State *L)
+{
+ Example *example = luaTHRD_checkudata(L, 1, fastnn_example_tname);
+ int idx = luaL_checkinteger(L, 2);
+ Matrix *m = Example_at(example, idx);
+ luaTHRD_pushudata(L, m, "nerv.CuMatrixFloat");
+ //luaT_pushudata(L, m, "nerv.CuMatrixFloat");
+ return 1;
+}
+
+static int example_pushback(lua_State *L)
+{
+ Example *example = luaTHRD_checkudata(L, 1, fastnn_example_tname);
+ Matrix *m = luaTHRD_checkudata(L, 2, "nerv.CuMatrixFloat");
+ //Matrix *m = luaT_checkudata(L, 2, "nerv.CuMatrixFloat");
+ Example_pushback(example, m);
+ return 0;
+}
+
+static int example_destroy(lua_State *L)
+{
+ Example *example = luaTHRD_checkudata(L, 1, fastnn_example_tname);
+ Example_destroy(example);
+ //printf("example_destroy ... end\n");
+ return 0;
+}
+
+
+static const struct luaL_Reg example_repo__ [] = {
+ {"new", example_repo_new},
+ {"newfromid", example_repo_newfromid},
+ {"__tostring", example_repo_tostring},
+ {"id", example_repo_id},
+ {"get_gpuid", example_repo_get_gpuid},
+ {"set_gpuid", example_repo_set_gpuid},
+ {"accept", example_repo_accept},
+ {"provide", example_repo_provide},
+ {"done", example_repo_done},
+ {"free", example_repo_destroy},
+ {NULL, NULL}
+};
+
+static const struct luaL_Reg example__ [] = {
+ {"new", example_new},
+ {"__tostring", example_tostring},
+ {"id", example_id},
+ {"size", example_size},
+ {"at", example_at},
+ {"pushback", example_pushback},
+ {"free", example_destroy},
+ {NULL, NULL}
+};
+
+void fastnn_init_example(lua_State *L)
+{
+ luaT_newmetatable(L, fastnn_example_tname, NULL, example_new, example_destroy, NULL);
+ luaL_register(L, NULL, example__);
+ lua_pop(L, 1);
+}
+
+void fastnn_init_example_repo(lua_State *L)
+{
+ luaT_newmetatable(L, fastnn_example_repo_tname, NULL, example_repo_new, example_repo_destroy, NULL);
+ luaL_register(L, NULL, example_repo__);
+ lua_pop(L, 1);
+}
+
+
+
+
diff --git a/fastnn/io/example.lua b/fastnn/io/example.lua
new file mode 100644
index 0000000..8808791
--- /dev/null
+++ b/fastnn/io/example.lua
@@ -0,0 +1,38 @@
+
+local T = require 'libthreads'
+local C = require 'libfastnn'
+
+local ExamplesRepo = nerv.class("fastnn.ExamplesRepo")
+local Example = nerv.class("fastnn.Example")
+
+fastnn.CExamplesRepo = C.CExamplesRepo
+fastnn.CExample = C.CExample
+
+function ExamplesRepo:__init(shareid)
+ if nil ~= shareid then
+ -- print(shareid)
+ self.repo = C.CExamplesRepo(shareid, true)
+ -- print(self.repo:__tostring())
+ end
+end
+
+
+function Example:PrepareData(data, global_transf, trains_id)
+ local example = fastnn.CExample()
+ if nil ~= data then
+ for id, cm in pairs(data) do
+ if trains_id[id] and nil ~= global_transf then
+ local tcm = nerv.speech_utils.normalize(cm, global_transf)
+ example:pushback(tcm)
+ else
+ example:pushback(cm)
+ end
+
+ end
+ end
+
+ return example
+end
+
+
+
diff --git a/fastnn/lib/ModelSync.h b/fastnn/lib/ModelSync.h
new file mode 100644
index 0000000..71216a0
--- /dev/null
+++ b/fastnn/lib/ModelSync.h
@@ -0,0 +1,119 @@
+
+#ifndef NERV_FASTNN_MODELSYNC_H
+#define NERV_FASTNN_MODELSYNC_H
+
+#define STRLEN 1024
+
+#include "../threads/lib/THThread.h"
+#include "matrix/matrix.h"
+#include "stdlib.h"
+#include "stdbool.h"
+
+typedef struct NnetParallelOptions_
+{
+ int num_threads;
+ int merge_size;
+ int num_merge;
+ int num_procs;
+ int threadid;
+ int myid;
+ int thread_level;
+ char merge_func[STRLEN];
+ char log_file[STRLEN];
+} NnetParallelOptions;
+
+
+typedef struct ModelSync_
+{
+ THMutex *model_mutex;
+ THMutex *state_mutex;
+ bool initialized_;
+ int dim_;
+ int pos_;
+ float *data_;
+ float *free_data_;
+ int refcount;
+ int threadcount;
+}ModelSync;
+
+ModelSync *ModelSync_new(void);
+ModelSync *ModelSync_newWithId(long id);
+int ModelSync_free(ModelSync *self);
+long ModelSync_id(ModelSync *self);
+int ModelSync_lockmodel(ModelSync *self);
+int ModelSync_unlockmodel(ModelSync *self);
+int ModelSync_lockstate(ModelSync *self);
+int ModelSync_unlockstate(ModelSync *self);
+int ModelSync_initBuffer(ModelSync *self);
+int ModelSync_weightfromd(ModelSync *self, Matrix *dm);
+int ModelSync_weighttod(ModelSync *self, Matrix *dm);
+int ModelSync_threadcount(ModelSync *self);
+void ModelSync_syncinc(ModelSync *self);
+void ModelSync_syncdec(ModelSync *self);
+
+typedef struct Xent_
+{
+ size_t frames_;
+ size_t correct_;
+ double loss_;
+ double entropy_;
+ int refcount;
+} Xent;
+
+Xent* Xent_new();
+Xent* Xent_newWithId(long id);
+Xent* Xent_newWithParm(size_t frames_, size_t correct_, double loss_, double entropy_);
+long Xent_id(Xent *xent);
+Xent* Xent_add(Xent *a, Xent *b);
+void Xent_free(Xent *xent);
+
+typedef struct Mse_
+{
+ size_t frames_;
+ double loss_;
+ int refcount;
+} Mse;
+
+Mse* Mse_new();
+Mse* Mse_newWithId(long id);
+Mse* Mse_newWithParm(size_t frames_, double loss_);
+long Mse_id(Mse *mse);
+Mse* Mse_add(Mse *a, Mse *b);
+void Mse_free(Mse *mse);
+
+typedef struct NnetUpdateState_
+{
+ int num_utter;
+ int num_nolabel;
+ int num_other_error;
+ long total_frames;
+ Xent xent;
+ Mse mse;
+} NnetUpdateState;
+
+typedef struct GlobalOption_
+{
+ int batch_size;
+ float lrate;
+ bool bp;
+ char tr_scp[STRLEN];
+ char cv_scp[STRLEN];
+ char transf[STRLEN];
+ char network[STRLEN];
+ int refcount;
+}GlobalOption;
+
+
+GlobalOption* GlobalOption_new();
+GlobalOption* GlobalOption_newWithParm(int batch_size, float lrate, bool bp, const char *tr_scp, const char *cv_scp, const char *transf, const char *network);
+GlobalOption* GlobalOption_newWithId(long id);
+long GlobalOption_id(GlobalOption *option);
+void GlobalOption_free(GlobalOption *option);
+
+
+
+
+#endif
+
+
+
diff --git a/fastnn/lib/modelsync.c b/fastnn/lib/modelsync.c
new file mode 100644
index 0000000..2b52752
--- /dev/null
+++ b/fastnn/lib/modelsync.c
@@ -0,0 +1,532 @@
+
+#include <stdio.h>
+#include <stdlib.h>
+
+#include <lua.h>
+#include <lualib.h>
+#include <luaT/luaT.h>
+
+
+
+#include "ModelSync.h"
+#include "../threads/lib/luaTHRD.h"
+
+const char *fastnn_model_sync_tname = "fastnn.CModelSync";
+const char *fastnn_xent_tname = "fastnn.CXent";
+const char *fastnn_mse_tname = "fastnn.CMse";
+const char *fastnn_global_option_tname = "fastnn.CGlobalOption";
+
+static int model_sync_new(lua_State *L)
+{
+ ModelSync *model_sync = NULL;
+ if(lua_gettop(L) == 0)
+ {
+ model_sync = ModelSync_new();
+ }
+ else if(lua_gettop(L) == 1)
+ {
+ long id = luaL_checklong(L, 1);
+ model_sync = ModelSync_newWithId(id);
+ }
+ else
+ luaL_error(L, "modelsync: modelsync new invalid arguments");
+ if(!model_sync)
+ luaL_error(L, "modelsync: modelsync new failed");
+
+ luaTHRD_pushudata(L, model_sync, fastnn_model_sync_tname);
+ return 1;
+}
+
+
+static int model_sync_tostring(lua_State *L)
+{
+ char str[STRLEN];
+ ModelSync *model_sync = luaTHRD_checkudata(L, 1, fastnn_model_sync_tname);
+ snprintf(str, STRLEN, "fastnn.modelsync <%lx>", ModelSync_id(model_sync));
+ lua_pushstring(L, str);
+ return 1;
+}
+
+static int model_sync_id(lua_State *L)
+{
+ ModelSync *model_sync = luaTHRD_checkudata(L, 1, fastnn_model_sync_tname);
+ lua_pushinteger(L, ModelSync_id(model_sync));
+ return 1;
+}
+
+static int model_sync_lockmodel(lua_State *L)
+{
+ ModelSync *model_sync = luaTHRD_checkudata(L, 1, fastnn_model_sync_tname);
+ if (ModelSync_lockmodel(model_sync))
+ luaL_error(L, "modelsync: model lock failed");
+ return 0;
+}
+
+static int model_sync_unlockmodel(lua_State *L)
+{
+ ModelSync *model_sync = luaTHRD_checkudata(L, 1, fastnn_model_sync_tname);
+ if (ModelSync_unlockmodel(model_sync))
+ luaL_error(L, "modelsync: model unlock failed");
+ return 0;
+}
+
+static int model_sync_lockstate(lua_State *L)
+{
+ ModelSync *model_sync = luaTHRD_checkudata(L, 1, fastnn_model_sync_tname);
+ if (ModelSync_lockstate(model_sync))
+ luaL_error(L, "modelsync: state lock failed");
+ return 0;
+}
+
+static int model_sync_unlockstate(lua_State *L)
+{
+ ModelSync *model_sync = luaTHRD_checkudata(L, 1, fastnn_model_sync_tname);
+ if (ModelSync_unlockstate(model_sync))
+ luaL_error(L, "modelsync: state unlock failed");
+ return 0;
+}
+
+static int model_sync_free(lua_State *L)
+{
+ ModelSync *model_sync = luaTHRD_checkudata(L, 1, fastnn_model_sync_tname);
+ ModelSync_free(model_sync);
+ return 0;
+}
+
+static int model_sync_initbuffer(lua_State *L)
+{
+ ModelSync *model_sync = luaTHRD_checkudata(L, 1, fastnn_model_sync_tname);
+ model_sync->dim_ = luaL_checkinteger(L, 2);
+ ModelSync_initBuffer(model_sync);
+ return 0;
+}
+
+static int model_sync_weightfromd(lua_State *L)
+{
+ ModelSync *model_sync = luaTHRD_checkudata(L, 1, fastnn_model_sync_tname);
+ Matrix *dm = luaT_checkudata(L, 2, "nerv.CuMatrixFloat");
+ ModelSync_weightfromd(model_sync, dm);
+ return 0;
+}
+
+static int model_sync_weighttod(lua_State *L)
+{
+ ModelSync *model_sync = luaTHRD_checkudata(L, 1, fastnn_model_sync_tname);
+ Matrix *dm = luaT_checkudata(L, 2, "nerv.CuMatrixFloat");
+ ModelSync_weighttod(model_sync, dm);
+ return 0;
+}
+
+static int model_sync_initialized(lua_State *L)
+{
+ ModelSync *model_sync = luaTHRD_checkudata(L, 1, fastnn_model_sync_tname);
+ lua_pushboolean(L, model_sync->initialized_);
+ return 1;
+}
+
+static int model_sync_setpos(lua_State *L)
+{
+ ModelSync *model_sync = luaTHRD_checkudata(L, 1, fastnn_model_sync_tname);
+ int pos = luaL_checkinteger(L, 2);
+ model_sync->pos_ = pos;
+ return 0;
+}
+
+static int model_sync_threadcount(lua_State *L)
+{
+ ModelSync *model_sync = luaTHRD_checkudata(L, 1, fastnn_model_sync_tname);
+ lua_pushinteger(L, ModelSync_threadcount(model_sync));
+ return 1;
+}
+
+static int model_sync_syncinc(lua_State *L)
+{
+ ModelSync *model_sync = luaTHRD_checkudata(L, 1, fastnn_model_sync_tname);
+ ModelSync_syncinc(model_sync);
+ return 0;
+}
+
+static int model_sync_syncdec(lua_State *L)
+{
+ ModelSync *model_sync = luaTHRD_checkudata(L, 1, fastnn_model_sync_tname);
+ ModelSync_syncdec(model_sync);
+ return 0;
+}
+//////////////////////////////////////////
+
+static int xent_new(lua_State *L)
+{
+ Xent *xent = NULL;
+ if(lua_gettop(L) == 0)
+ {
+ xent = Xent_new();
+ }
+ else if(lua_gettop(L) == 1)
+ {
+ long id = luaL_checklong(L, 1);
+ xent = Xent_newWithId(id);
+ }
+ else if(lua_gettop(L) == 4)
+ {
+ size_t frames_, correct_;
+ double loss_, entropy_ ;
+ frames_ = luaL_checkinteger(L, 1);
+ correct_ = luaL_checkinteger(L, 2);
+ loss_ = luaL_checknumber(L, 3);
+ entropy_ = luaL_checknumber(L, 4);
+ xent = Xent_newWithParm(frames_, correct_, loss_, entropy_);
+ }
+ else
+ luaL_error(L, "xent: xent new invalid arguments");
+ if(!xent)
+ luaL_error(L, "xent: xent new failed");
+
+ luaTHRD_pushudata(L, xent, fastnn_xent_tname);
+ return 1;
+}
+
+static int xent_tostring(lua_State *L)
+{
+ char str[STRLEN];
+ Xent *xent = luaTHRD_checkudata(L, 1, fastnn_xent_tname);
+ snprintf(str, STRLEN, "fastnn.xent <%lx>", Xent_id(xent));
+ lua_pushstring(L, str);
+ return 1;
+}
+
+static int xent_totalframes(lua_State *L)
+{
+ Xent *xent = luaTHRD_checkudata(L, 1, fastnn_xent_tname);
+ lua_pushinteger(L, xent->frames_);
+ return 1;
+}
+
+static int xent_correct(lua_State *L)
+{
+ Xent *xent = luaTHRD_checkudata(L, 1, fastnn_xent_tname);
+ lua_pushinteger(L, xent->correct_);
+ return 1;
+}
+
+static int xent_loss(lua_State *L)
+{
+ Xent *xent = luaTHRD_checkudata(L, 1, fastnn_xent_tname);
+ lua_pushnumber(L, xent->loss_);
+ return 1;
+}
+
+static int xent_entropy(lua_State *L)
+{
+ Xent *xent = luaTHRD_checkudata(L, 1, fastnn_xent_tname);
+ lua_pushnumber(L, xent->entropy_);
+ return 1;
+}
+
+static int xent_id(lua_State *L)
+{
+ Xent *xent = luaTHRD_checkudata(L, 1, fastnn_xent_tname);
+ lua_pushinteger(L, Xent_id(xent));
+ return 1;
+}
+
+static int xent_free(lua_State *L)
+{
+ Xent *xent = luaTHRD_checkudata(L, 1, fastnn_xent_tname);
+ Xent_free(xent);
+ return 0;
+}
+
+static int xent_add(lua_State *L)
+{
+ Xent *a = luaTHRD_checkudata(L, 1, fastnn_xent_tname);
+ Xent *b = luaTHRD_checkudata(L, 2, fastnn_xent_tname);
+ Xent_add(a, b);
+ return 0;
+}
+
+/////////////////////////////////////////////
+
+static int mse_new(lua_State *L)
+{
+ Mse *mse = NULL;
+ if(lua_gettop(L) == 0)
+ {
+ mse = Mse_new();
+ }
+ else if(lua_gettop(L) == 1)
+ {
+ long id = luaL_checklong(L, 1);
+ mse = Mse_newWithId(id);
+ }
+ else if(lua_gettop(L) == 2)
+ {
+ size_t frames_;
+ double loss_;
+ frames_ = luaL_checkinteger(L, 1);
+ loss_ = luaL_checknumber(L, 2);
+ mse = Mse_newWithParm(frames_, loss_);
+ }
+ else
+ luaL_error(L, "mse: mse new invalid arguments");
+ if(!mse)
+ luaL_error(L, "mse: mse new failed");
+
+ luaTHRD_pushudata(L, mse, fastnn_mse_tname);
+ return 1;
+
+}
+
+static int mse_tostring(lua_State *L)
+{
+ char str[STRLEN];
+ Mse *mse = luaTHRD_checkudata(L, 1, fastnn_mse_tname);
+ snprintf(str, STRLEN, "fastnn.mse <%lx>", Mse_id(mse));
+ lua_pushstring(L, str);
+ return 1;
+}
+
+static int mse_totalframes(lua_State *L)
+{
+ Mse *mse = luaTHRD_checkudata(L, 1, fastnn_mse_tname);
+ lua_pushinteger(L, mse->frames_);
+ return 1;
+}
+
+static int mse_loss(lua_State *L)
+{
+ Mse *mse = luaTHRD_checkudata(L, 1, fastnn_mse_tname);
+ lua_pushnumber(L, mse->loss_);
+ return 1;
+}
+
+static int mse_id(lua_State *L)
+{
+ Mse *mse = luaTHRD_checkudata(L, 1, fastnn_mse_tname);
+ lua_pushinteger(L, Mse_id(mse));
+ return 1;
+}
+
+static int mse_free(lua_State *L)
+{
+ Mse *mse = luaTHRD_checkudata(L, 1, fastnn_mse_tname);
+ Mse_free(mse);
+ return 0;
+}
+
+static int mse_add(lua_State *L)
+{
+ Mse *a = luaTHRD_checkudata(L, 1, fastnn_mse_tname);
+ Mse *b = luaTHRD_checkudata(L, 2, fastnn_mse_tname);
+ Mse_add(a, b);
+ return 0;
+}
+/////////////////////////////////////////////
+
+static int global_option_new(lua_State *L)
+{
+ GlobalOption *global_option = NULL;
+ if(lua_gettop(L) == 0)
+ {
+ global_option = GlobalOption_new();
+ }
+ else if(lua_gettop(L) == 1)
+ {
+ long id = luaL_checklong(L, 1);
+ global_option = GlobalOption_newWithId(id);
+ }
+ else if(lua_gettop(L) > 3)
+ {
+ int batch_size = luaL_checkinteger(L, 1);
+ float lrate = luaL_checknumber(L, 2);
+ bool bp = lua_toboolean(L, 3);
+ const char *tr_scp = lua_tostring(L, 4);
+ const char *cv_scp = lua_tostring(L, 5);
+ const char *transf = lua_tostring(L, 6);
+ const char *network = lua_tostring(L, 7);
+ global_option = GlobalOption_newWithParm(batch_size, lrate, bp, tr_scp, cv_scp, transf, network);
+ }
+ else
+ luaL_error(L, "global_option: global_option new invalid arguments");
+ if(!global_option)
+ luaL_error(L, "global_option: global_option new failed");
+
+ luaTHRD_pushudata(L, global_option, fastnn_global_option_tname);
+ return 1;
+
+}
+
+static int global_option_tostring(lua_State *L)
+{
+ char str[STRLEN];
+ GlobalOption *global_option = luaTHRD_checkudata(L, 1, fastnn_global_option_tname);
+ snprintf(str, STRLEN, "fastnn.global_option <%lx>", GlobalOption_id(global_option));
+ lua_pushstring(L, str);
+ return 1;
+}
+
+static int global_option_id(lua_State *L)
+{
+ GlobalOption *global_option = luaTHRD_checkudata(L, 1, fastnn_global_option_tname);
+ lua_pushinteger(L, GlobalOption_id(global_option));
+ return 1;
+}
+
+static int global_option_free(lua_State *L)
+{
+ GlobalOption *global_option = luaTHRD_checkudata(L, 1, fastnn_global_option_tname);
+ GlobalOption_free(global_option);
+ return 0;
+}
+
+static int global_option_batch_size(lua_State *L)
+{
+ GlobalOption *global_option = luaTHRD_checkudata(L, 1, fastnn_global_option_tname);
+ lua_pushinteger(L, global_option->batch_size);
+ return 1;
+}
+
+static int global_option_lrate(lua_State *L)
+{
+ GlobalOption *global_option = luaTHRD_checkudata(L, 1, fastnn_global_option_tname);
+ lua_pushnumber(L, global_option->lrate);
+ return 1;
+}
+
+static int global_option_bp(lua_State *L)
+{
+ GlobalOption *global_option = luaTHRD_checkudata(L, 1, fastnn_global_option_tname);
+ lua_pushboolean(L, global_option->bp);
+ return 1;
+}
+
+static int global_option_tr_scp(lua_State *L)
+{
+ GlobalOption *global_option = luaTHRD_checkudata(L, 1, fastnn_global_option_tname);
+ lua_pushstring(L, global_option->tr_scp);
+ return 1;
+}
+
+static int global_option_cv_scp(lua_State *L)
+{
+ GlobalOption *global_option = luaTHRD_checkudata(L, 1, fastnn_global_option_tname);
+ lua_pushstring(L, global_option->cv_scp);
+ return 1;
+}
+
+static int global_option_transf(lua_State *L)
+{
+ GlobalOption *global_option = luaTHRD_checkudata(L, 1, fastnn_global_option_tname);
+ lua_pushstring(L, global_option->transf);
+ return 1;
+}
+
+static int global_option_network(lua_State *L)
+{
+ GlobalOption *global_option = luaTHRD_checkudata(L, 1, fastnn_global_option_tname);
+ lua_pushstring(L, global_option->network);
+ return 1;
+}
+
+
+//////////////////////////////////////////////
+
+static const struct luaL_Reg model_sync__ [] = {
+ {"new", model_sync_new},
+ {"__tostring", model_sync_tostring},
+ {"id", model_sync_id},
+ {"lockmodel", model_sync_lockmodel},
+ {"unlockmodel", model_sync_unlockmodel},
+ {"lockstate", model_sync_lockstate},
+ {"unlockstate", model_sync_unlockstate},
+ {"initbuffer", model_sync_initbuffer},
+ {"setpos", model_sync_setpos},
+ {"initialized", model_sync_initialized},
+ {"weightfromd", model_sync_weightfromd},
+ {"weighttod", model_sync_weighttod},
+ {"threadcount", model_sync_threadcount},
+ {"syncinc", model_sync_syncinc},
+ {"syncdec", model_sync_syncdec},
+ {"threadcount", model_sync_threadcount},
+ {"free", model_sync_free},
+ {NULL, NULL}
+};
+
+
+static const struct luaL_Reg xent__ [] = {
+ {"new", xent_new},
+ {"__tostring", xent_tostring},
+ {"id", xent_id},
+ {"totalframes", xent_totalframes},
+ {"correct", xent_correct},
+ {"loss", xent_loss},
+ {"entropy", xent_entropy},
+ {"add", xent_add},
+ {"free", xent_free},
+ {NULL, NULL}
+};
+
+static const struct luaL_Reg mse__ [] = {
+ {"new", mse_new},
+ {"__tostring", mse_tostring},
+ {"id", mse_id},
+ {"totalframes", mse_totalframes},
+ {"loss", mse_loss},
+ {"add", mse_add},
+ {"free", mse_free},
+ {NULL, NULL}
+};
+
+
+static const struct luaL_Reg global_option__ [] = {
+ {"new", global_option_new},
+ {"__tostring", global_option_tostring},
+ {"id", global_option_id},
+ {"batch_size", global_option_batch_size},
+ {"lrate", global_option_lrate},
+ {"bp", global_option_bp},
+ {"tr_scp", global_option_tr_scp},
+ {"cv_scp", global_option_cv_scp},
+ {"transf", global_option_transf},
+ {"network", global_option_network},
+ {"free", global_option_free},
+ {NULL, NULL}
+};
+
+void fastnn_init_modelsync(lua_State *L)
+{
+ luaT_newmetatable(L, fastnn_model_sync_tname, NULL, model_sync_new, NULL, NULL);
+ luaL_register(L, NULL, model_sync__);
+ lua_pop(L, 1);
+
+ luaT_newmetatable(L, fastnn_xent_tname, NULL, xent_new, xent_free, NULL);
+ luaL_register(L, NULL, xent__);
+ lua_pop(L, 1);
+
+ luaT_newmetatable(L, fastnn_mse_tname, NULL, mse_new, mse_free, NULL);
+ luaL_register(L, NULL, mse__);
+ lua_pop(L, 1);
+
+ luaT_newmetatable(L, fastnn_global_option_tname, NULL, global_option_new, global_option_free, NULL);
+ luaL_register(L, NULL, global_option__);
+ lua_pop(L, 1);
+ /*
+ printf("%s %lx\n", model_sync__[13].name, model_sync__[13].func);
+
+ if(!luaL_newmetatable(L, fastnn_model_sync_tname))
+ luaL_error(L, "fastnn: fastnn.modelsync type already exists");
+ luaL_setfuncs(L, model_sync__, 0);
+ lua_pushstring(L, "__index");
+ lua_pushvalue(L, -2);
+ lua_rawset(L, -3);
+ lua_pop(L, 1);
+
+ printf("%s %lx\n", model_sync__[13].name, model_sync__[13].func);
+
+ lua_pushstring(L, "modelsync");
+ luaTHRD_pushctortable(L, model_sync_new, fastnn_model_sync_tname);
+ lua_rawset(L, -3);
+ printf("%s %lx\n", model_sync__[13].name, model_sync__[13].func);
+ */
+}
+
+
diff --git a/fastnn/lib/modelsync.lua b/fastnn/lib/modelsync.lua
new file mode 100644
index 0000000..a247562
--- /dev/null
+++ b/fastnn/lib/modelsync.lua
@@ -0,0 +1,107 @@
+
+local C = require 'libfastnn'
+local T = require 'libthreads'
+
+local ModelSync = nerv.class("fastnn.ModelSync")
+
+fastnn.CModelSync = C.CModelSync
+fastnn.Thread = T.Thread
+
+
+function ModelSync:__init(shareid)
+ self.modelsync = fastnn.CModelSync(shareid)
+-- print(self.modelsync.initbuffer)
+ --print(self.modelsync.setpos)
+ --print(self.modelsync.initialized)
+ --print(self.modelsync.weightfromd)
+-- print(self.modelsync.weighttod)
+-- print(self.modelsync.aaaa)
+-- print(self.modelsync.bbbb)
+-- print(self.modelsync.cccc)
+end
+
+function ModelSync:GetDim(nnet)
+
+ local repo = nnet:get_params()
+ local params = repo.params
+ local dim = 0
+ for pid, ref in pairs(params) do
+ if nerv.is_type(ref.trans, "nerv.Matrix") then
+ dim = dim + ref.trans:nrow() * ref.trans:nstride()
+ end
+ end
+
+ return dim
+end
+
+
+function ModelSync:Initialize(nnet)
+
+ self:LockModel()
+
+ if not self.modelsync:initialized() then
+ dim = self:GetDim(nnet)
+ self.modelsync:initbuffer(dim)
+ self:WeightFromD(nnet)
+ end
+
+ self:UnLockModel()
+end
+
+function ModelSync:WeightFromD(nnet)
+ local repo = nnet:get_params()
+ local params = repo.params
+ self.modelsync:setpos(0)
+ for pid, ref in pairs(params) do
+ if nerv.is_type(ref.trans, "nerv.Matrix") then
+ self.modelsync:weightfromd(ref.trans)
+ end
+ end
+end
+
+function ModelSync:WeightToD(nnet)
+ local repo = nnet:get_params()
+ local params = repo.params
+ self.modelsync:setpos(0)
+ for pid, ref in pairs(params) do
+ if nerv.is_type(ref.trans, "nerv.Matrix") then
+ self.modelsync:weighttod(ref.trans)
+ end
+ end
+end
+
+function ModelSync:LockState()
+ self.modelsync:lockstate()
+end
+
+function ModelSync:UnLockState()
+ self.modelsync:unlockstate()
+end
+
+
+function ModelSync:LockModel()
+ self.modelsync:lockmodel()
+end
+
+
+function ModelSync:UnLockModel()
+ self.modelsync:unlockmodel()
+end
+
+function ModelSync:Id()
+ return self.modelsync:id()
+end
+
+function ModelSync:ThreadCount()
+ return self.modelsync:threadcount()
+end
+
+function ModelSync:SyncInc()
+ return self.modelsync:syncinc()
+end
+
+function ModelSync:SyncDec()
+ return self.modelsync:syncdec()
+end
+
+
diff --git a/fastnn/test.lua b/fastnn/test.lua
new file mode 100644
index 0000000..3b56eb1
--- /dev/null
+++ b/fastnn/test.lua
@@ -0,0 +1,57 @@
+require 'fastnn'
+
+function build_trainer(ifname)
+ local param_repo = nerv.ParamRepo()
+ param_repo:import(ifname, nil, gconf)
+ local sublayer_repo = make_sublayer_repo(param_repo)
+ local layer_repo = make_layer_repo(sublayer_repo, param_repo)
+ local nnet = get_network(layer_repo)
+ local input_order = get_input_order()
+
+
+ local iterative_trainer = function (prefix, scp_file, bp)
+
+ -- build buffer
+ local buffer = make_buffer(make_readers(scp_file, layer_repo))
+
+ --[[local control = fastnn.modelsync();
+ local lua_control = fastnn.ModelSync(control:id())
+ print(control:__tostring())
+ print(lua_control:GetDim(nnet))
+ lua_control:Initialize(nnet)
+ lua_control:WeightToD(nnet)
+ lua_control:WeightToD(nnet)
+ ]]
+
+ local example_repo = fastnn.CExamplesRepo(128, false)
+ -- print(example_repo)
+ local share_repo = fastnn.CExamplesRepo(example_repo:id(), true)
+ feat_id = get_feat_id()
+
+ local t = 1;
+ for data in buffer.get_data, buffer do
+ local example = fastnn.Example.PrepareData(data, layer_repo.global_transf, feat_id)
+ print(example)
+ share_repo:accept(example)
+ end
+
+ end
+ return iterative_trainer
+end
+
+
+dofile(arg[1])
+start_halving_inc = 0.5
+halving_factor = 0.6
+end_halving_inc = 0.1
+min_iter = 1
+max_iter = 20
+min_halving = 5
+gconf.batch_size = 256
+gconf.buffer_size = 81920
+
+local pf0 = gconf.initialized_param
+local trainer = build_trainer(pf0)
+
+local accu_best = trainer(nil, gconf.cv_scp, false)
+
diff --git a/fastnn/threads/Makefile b/fastnn/threads/Makefile
new file mode 100644
index 0000000..17958f9
--- /dev/null
+++ b/fastnn/threads/Makefile
@@ -0,0 +1,42 @@
+.PHONY: build install clean
+SHELL := /bin/bash
+BUILD_DIR := $(CURDIR)/build
+INC_PATH := $(LUA_BINDIR)/../include/nerv/luaT
+LUA_DIR = $(INST_LUADIR)/threads
+LIB_PATH := $(LUA_BINDIR)/../lib
+
+OBJ_DIR := $(BUILD_DIR)/objs
+SUBDIR := lib test
+
+
+OBJS := lib/init.o lib/threads.o lib/THThread.o
+LIBS := $(INST_LIBDIR)/libthreads.so
+LUA_LIBS := init.lua threads.lua test/test-threads.lua
+INCLUDE := -I $(LUA_INCDIR) -I $(INC_PATH) -DLUA_USE_APICHECK -DUSE_PTHREAD_THREADS
+
+
+OBJS := $(addprefix $(OBJ_DIR)/,$(OBJS))
+OBJ_SUBDIR := $(addprefix $(OBJ_DIR)/,$(SUBDIR))
+LUA_SUBDIR := $(addprefix $(LUA_DIR)/,$(SUBDIR))
+LUA_LIBS := $(addprefix $(LUA_DIR)/,$(LUA_LIBS))
+
+CFLAGS := -Wall -Wextra -O0
+
+build: $(OBJ_DIR) $(OBJ_SUBDIR) $(OBJS)
+install: $(LUA_DIR) $(LUA_SUBDIR) $(LIBS) $(LUA_LIBS)
+
+$(OBJ_DIR) $(LUA_DIR) $(OBJ_SUBDIR) $(LUA_SUBDIR):
+ -mkdir -p $@
+$(LUA_DIR)/%.lua: %.lua
+ cp $< $@
+
+$(OBJ_DIR)/%.o: %.c $(patsubst /%.o,/%.c,$@)
+ gcc -c -o $@ $< $(INCLUDE) -fPIC $(CFLAGS)
+
+$(LIBS): $(OBJS)
+ gcc -shared -o $@ $^ $(LDFLAGS) -Wl,-rpath=$(LIB_PATH) -L$(LIB_PATH) -lnervcore -lluaT -lpthread
+
+clean:
+ -rm -rf $(OBJ_DIR)
+
+
diff --git a/fastnn/threads/init.lua b/fastnn/threads/init.lua
new file mode 100644
index 0000000..6cd3679
--- /dev/null
+++ b/fastnn/threads/init.lua
@@ -0,0 +1,14 @@
+threads = {}
+
+local C = require 'libthreads'
+
+threads.Thread = C.Thread
+threads.Mutex = C.Mutex
+threads.Condition = C.Condition
+threads.Semaphore = C.Semaphore
+--threads.Threads = require 'threads.threads'
+
+-- only for backward compatibility (boo)
+-- setmetatable(threads, getmetatable(threads.Threads))
+
+-- return threads
diff --git a/fastnn/threads/lib/THThread.c b/fastnn/threads/lib/THThread.c
new file mode 100644
index 0000000..7abc044
--- /dev/null
+++ b/fastnn/threads/lib/THThread.c
@@ -0,0 +1,349 @@
+#ifndef TH_THREAD_INC
+#define TH_THREAD_INC
+
+#include <stdlib.h>
+#include <string.h>
+
+/*#include "TH.h" */
+#include "THThread.h"
+
+#if defined(USE_PTHREAD_THREADS)
+#include <pthread.h>
+
+#elif defined(USE_WIN32_THREADS)
+
+/* very basic emulation to suit our needs */
+
+#include <process.h>
+#include <windows.h>
+
+typedef HANDLE pthread_t;
+typedef DWORD pthread_attr_t;
+typedef HANDLE pthread_mutex_t;
+typedef HANDLE pthread_cond_t;
+
+static int pthread_create(pthread_t *restrict thread,
+ const pthread_attr_t *restrict attr, void *(*start_routine)(void *),
+ void *restrict arg)
+{
+ *thread = (HANDLE)_beginthreadex(NULL, 0, (THREAD_FUNCTION)start_routine, arg, 0, NULL);
+ return (int)(*thread == NULL);
+}
+
+static int pthread_join(pthread_t thread, void **value_ptr)
+{
+ return ((WaitForSingleObject((thread), INFINITE) != WAIT_OBJECT_0) || !CloseHandle(thread));
+}
+
+static int pthread_mutex_init(pthread_mutex_t *restrict mutex,
+ const pthread_mutexattr_t *restrict attr)
+{
+ *mutex = CreateMutex(NULL, FALSE, NULL);
+ return (int)(*mutex == NULL);
+}
+
+static int pthread_mutex_lock(pthread_mutex_t *mutex)
+{
+ return WaitForSingleObject(*mutex, INFINITE) == 0;
+}
+
+static int pthread_mutex_unlock(pthread_mutex_t *mutex)
+{
+ return ReleaseMutex(*mutex) == 0;
+}
+
+static int pthread_mutex_destroy(pthread_mutex_t *mutex)
+{
+ return CloseHandle(*mutex) == 0;
+}
+
+static int pthread_cond_init(pthread_cond_t *restrict cond,
+ const pthread_condattr_t *restrict attr)
+{
+ *cond = CreateEvent(NULL, FALSE, FALSE, NULL);
+ return (int)(*cond == NULL);
+}
+
+static int pthread_cond_wait(pthread_cond_t *restrict cond,
+ pthread_mutex_t *restrict mutex)
+{
+ SignalObjectAndWait(*mutex, *cond, INFINITE, FALSE);
+ return WaitForSingleObject(*mutex, INFINITE) == 0;
+}
+
+static int pthread_cond_destroy(pthread_cond_t *cond)
+{
+ return CloseHandle(*cond) == 0;
+}
+
+int pthread_cond_signal(pthread_cond_t *cond)
+{
+ return SetEvent(*cond) == 0;
+}
+
+#else
+#error no thread system available
+#endif
+
+typedef struct THThread_ {
+ pthread_t id;
+ int (*func)(void*);
+ void* data;
+ int status;
+} THThread;
+
+typedef struct THMutex_{
+ pthread_mutex_t id;
+ int refcount;
+} THMutex;
+
+typedef struct THCondition_ {
+ pthread_cond_t id;
+ int refcount;
+} THCondition;
+
+typedef struct THSemaphore_ {
+ pthread_mutex_t mutex_;
+ pthread_cond_t cond_;
+ int counter_;
+ int refcount;
+}THSemaphore;
+
+static void* thread_closure(void *data)
+{
+ THThread *thread = data;
+ thread->status = thread->func(thread->data);
+ return NULL;
+}
+
+THThread* THThread_new(int (*func)(void*), void *data)
+{
+ THThread *self = malloc(sizeof(THThread));
+ self->func = func;
+ self->data = data;
+ self->status = 0;
+ if(!self)
+ return NULL;
+ if(pthread_create(&self->id, NULL, thread_closure, self)) {
+ free(self);
+ return NULL;
+ }
+ return self;
+}
+
+long THThread_id(THThread *self)
+{
+ return (long)(self);
+}
+
+int THThread_free(THThread *self)
+{
+ int status = 1;
+ if(self) {
+ if(pthread_join(self->id, NULL))
+ return 1;
+ status = self->status;
+ free(self);
+ self = NULL;
+ }
+ return status;
+}
+
+THMutex* THMutex_new(void)
+{
+ THMutex *self = malloc(sizeof(THMutex));
+ if(!self)
+ return NULL;
+ if(pthread_mutex_init(&self->id, NULL) != 0) {
+ free(self);
+ return NULL;
+ }
+ self->refcount = 1;
+ return self;
+}
+
+THMutex* THMutex_newWithId(long id)
+{
+ THMutex *self = (THMutex*)id;
+ __sync_fetch_and_add(&self->refcount, 1);
+ return self;
+}
+
+long THMutex_id(THMutex *self)
+{
+ return (long)(self);
+}
+
+int THMutex_lock(THMutex *self)
+{
+ if(pthread_mutex_lock(&self->id) != 0)
+ return 1;
+ return 0;
+}
+
+int THMutex_trylock(THMutex *self)
+{
+ if (pthread_mutex_trylock(&self->id) != 0)
+ return 1;
+ return 0;
+}
+
+int THMutex_unlock(THMutex *self)
+{
+ if(pthread_mutex_unlock(&self->id) != 0)
+ return 1;
+ return 0;
+}
+
+void THMutex_free(THMutex *self)
+{
+ if(self) {
+ if(__sync_fetch_and_add(&self->refcount, -1) == 1) {
+ pthread_mutex_destroy(&self->id);
+ free(self);
+ self = NULL;
+ }
+ }
+}
+
+THCondition* THCondition_new(void)
+{
+ THCondition *self = malloc(sizeof(THCondition));
+ if(!self)
+ return NULL;
+ if(pthread_cond_init(&self->id, NULL)) {
+ free(self);
+ return NULL;
+ }
+ self->refcount = 1;
+ return self;
+}
+
+THCondition* THCondition_newWithId(long id)
+{
+ THCondition *self = (THCondition*)id;
+ __sync_fetch_and_add(&self->refcount, 1);
+ return self;
+}
+
+long THCondition_id(THCondition *self)
+{
+ return (long)(self);
+}
+
+int THCondition_signal(THCondition *self)
+{
+ if(pthread_cond_signal(&self->id))
+ return 1;
+ return 0;
+}
+
+int THCondition_wait(THCondition *self, THMutex *mutex)
+{
+ if(pthread_cond_wait(&self->id, &mutex->id))
+ return 1;
+ return 0;
+}
+
+void THCondition_free(THCondition *self)
+{
+ if(self) {
+ if(__sync_fetch_and_add(&self->refcount, -1) == 1) {
+ pthread_cond_destroy(&self->id);
+ free(self);
+ self = NULL;
+ }
+ }
+}
+
+THSemaphore* THSemaphore_new(int initValue)
+{
+ THSemaphore *self = malloc(sizeof(THSemaphore));
+ if(!self)
+ return NULL;
+ self->counter_ = initValue;
+ if(pthread_mutex_init(&self->mutex_, NULL) != 0)
+ {
+ free(self);
+ return NULL;
+ }
+
+ if(pthread_cond_init(&self->cond_, NULL) != 0)
+ {
+ free(self);
+ return NULL;
+ }
+
+ self->refcount = 1;
+
+ return self;
+}
+
+THSemaphore* THSemaphore_newWithId(long id)
+{
+ THSemaphore *self = (THSemaphore*)id;
+ __sync_fetch_and_add(&self->refcount, 1);
+ return self;
+}
+
+long THSemaphore_id(THSemaphore *self)
+{
+ return (long)(self);
+}
+
+int THSemaphore_wait(THSemaphore *self)
+{
+ int ret = 0;
+ ret |= pthread_mutex_lock(&self->mutex_);
+
+ while (self->counter_ <= 0)
+ ret |= pthread_cond_wait(&self->cond_, &self->mutex_);
+
+ self->counter_--;
+ ret |= pthread_mutex_unlock(&self->mutex_);
+
+ return ret;
+
+}
+
+int THSemaphore_signal(THSemaphore *self)
+{
+ int ret = 0;
+ ret |= pthread_mutex_lock(&self->mutex_);
+ self->counter_++;
+ ret |= pthread_cond_signal(&self->cond_);
+ ret |= pthread_mutex_unlock(&self->mutex_);
+
+ return ret;
+}
+
+int THSemaphore_trywait(THSemaphore *self)
+{
+ int ret = 0;
+ int try_wait_succeeded = 1;
+ ret |= pthread_mutex_lock(&self->mutex_);
+ if (self->counter_ > 0) {
+ self->counter_--;
+ try_wait_succeeded = 0;
+ }
+ ret |= pthread_mutex_unlock(&self->mutex_);
+ return try_wait_succeeded;
+}
+
+void THSemaphore_free(THSemaphore *self)
+{
+ if(self)
+ {
+ if(__sync_fetch_and_add(&self->refcount, -1) == 1)
+ {
+ pthread_mutex_destroy(&self->mutex_);
+ pthread_cond_destroy(&self->cond_);
+ free(self);
+ self = NULL;
+ }
+ }
+ }
+
+
+
+#endif
diff --git a/fastnn/threads/lib/THThread.h b/fastnn/threads/lib/THThread.h
new file mode 100644
index 0000000..75ba417
--- /dev/null
+++ b/fastnn/threads/lib/THThread.h
@@ -0,0 +1,36 @@
+#ifndef TH_THREAD_INC
+#define TH_THREAD_INC
+
+typedef struct THThread_ THThread;
+typedef struct THMutex_ THMutex;
+typedef struct THCondition_ THCondition;
+typedef struct THSemaphore_ THSemaphore;
+
+THThread* THThread_new(int (*closure)(void*), void *data);
+long THThread_id(THThread *self);
+int THThread_free(THThread *self);
+
+THMutex* THMutex_new(void);
+THMutex* THMutex_newWithId(long id);
+long THMutex_id(THMutex *self);
+int THMutex_lock(THMutex *self);
+int THMutex_unlock(THMutex *self);
+int THMutex_trylock(THMutex *self);
+void THMutex_free(THMutex *self);
+
+THCondition* THCondition_new(void);
+THCondition* THCondition_newWithId(long id);
+long THCondition_id(THCondition *self);
+int THCondition_signal(THCondition *self);
+int THCondition_wait(THCondition *self, THMutex *mutex);
+void THCondition_free(THCondition *self);
+
+THSemaphore* THSemaphore_new(int initValue);
+THSemaphore* THSemaphore_newWithId(long id);
+long THSemaphore_id(THSemaphore *self);
+int THSemaphore_signal(THSemaphore *self);
+int THSemaphore_wait(THSemaphore *self);
+int THSemaphore_trywait(THSemaphore *self);
+void THSemaphore_free(THSemaphore *self);
+
+#endif
diff --git a/fastnn/threads/lib/init.c b/fastnn/threads/lib/init.c
new file mode 100644
index 0000000..4042189
--- /dev/null
+++ b/fastnn/threads/lib/init.c
@@ -0,0 +1,27 @@
+#include <lua.h>
+#include <lauxlib.h>
+
+#if LUA_VERSION_NUM == 501
+static void luaL_setfuncs(lua_State *L, const luaL_Reg *l, int nup)
+{
+ luaL_checkstack(L, nup+1, "too many upvalues");
+ for (; l->name != NULL; l++) { /* fill the table with given functions */
+ int i;
+ lua_pushstring(L, l->name);
+ for (i = 0; i < nup; i++) /* copy upvalues to the top */
+ lua_pushvalue(L, -(nup+1));
+ lua_pushcclosure(L, l->func, nup); /* closure with those upvalues */
+ lua_settable(L, -(nup + 3));
+ }
+ lua_pop(L, nup); /* remove upvalues */
+}
+#endif
+
+#include "threads.c"
+
+int luaopen_libthreads(lua_State *L)
+{
+ lua_newtable(L);
+ thread_init_pkg(L);
+ return 1;
+}
diff --git a/fastnn/threads/lib/luaTHRD.h b/fastnn/threads/lib/luaTHRD.h
new file mode 100644
index 0000000..3760bdb
--- /dev/null
+++ b/fastnn/threads/lib/luaTHRD.h
@@ -0,0 +1,84 @@
+#ifndef LUA_THRD_INC
+#define LUA_THRD_INC
+
+static int luaTHRD_pushudata(lua_State *L, void *ptr, const char* typename)
+{
+ void **udata = lua_newuserdata(L, sizeof(void*));
+ if(udata) {
+ *udata = ptr;
+ luaL_getmetatable(L, typename);
+ lua_setmetatable(L, -2);
+ return 1;
+ }
+ return 0;
+}
+
+static void *luaTHRD_checkudata(lua_State *L, int narg, const char *typename)
+{
+ void **udata = luaL_checkudata(L, narg, typename);
+ if(udata)
+ return *udata;
+ else
+ return NULL;
+}
+
+static void *luaTHRD_toudata(lua_State *L, int narg, const char *typename)
+{
+ void **udata = lua_touserdata(L, narg);
+ if(udata) {
+ if(lua_getmetatable(L, -1)) {
+ luaL_getmetatable(L, typename);
+ if(lua_equal(L, -1, -2)) {
+ lua_pop(L, 2);
+ return *udata;
+ }
+ else {
+ lua_pop(L, 2);
+ return NULL;
+ }
+ }
+ else
+ return NULL;
+ }
+ else
+ return NULL;
+}
+
+static int luaTHRD_ctor(lua_State *L)
+{
+ if(!lua_istable(L, 1)) /* dummy ctor table */
+ luaL_error(L, "ctor: table expected");
+ lua_getmetatable(L, 1);
+ lua_remove(L, 1); /* dummy ctor table */
+ if(!lua_istable(L, -1))
+ luaL_error(L, "ctor: no metatable found");
+ lua_pushstring(L, "__new");
+ lua_rawget(L, -2);
+ lua_remove(L, -2); /* forget about metatable */
+ if(!lua_isfunction(L, -1))
+ luaL_error(L, "ctor: __new appears to be not a function");
+ lua_insert(L, 1); /* ctor first, arguments follow */
+ lua_call(L, lua_gettop(L)-1, LUA_MULTRET);
+ return lua_gettop(L);
+}
+
+static void luaTHRD_pushctortable(lua_State *L, lua_CFunction ctor, const char* typename)
+{
+ lua_newtable(L); /* empty useless dude */
+ lua_newtable(L); /* metatable of the dude */
+ lua_pushstring(L, "__index");
+ luaL_getmetatable(L, typename);
+ lua_rawset(L, -3);
+ lua_pushstring(L, "__newindex");
+ luaL_getmetatable(L, typename);
+ lua_rawset(L, -3);
+ lua_pushstring(L, "__new"); /* __call will look into there */
+ lua_pushcfunction(L, ctor);
+ lua_rawset(L, -3);
+ lua_pushstring(L, "__call"); /* pop the table and calls __new */
+ lua_pushcfunction(L, luaTHRD_ctor);
+ lua_rawset(L, -3);
+ lua_setmetatable(L, -2);
+}
+
+#endif
diff --git a/fastnn/threads/lib/threads.c b/fastnn/threads/lib/threads.c
new file mode 100644
index 0000000..bf2a5ec
--- /dev/null
+++ b/fastnn/threads/lib/threads.c
@@ -0,0 +1,355 @@
+#include <stdio.h>
+#include <stdlib.h>
+#include <luaT.h>
+#include <string.h>
+
+#include "THThread.h"
+#include "luaTHRD.h"
+
+#include <lua.h>
+#include <lualib.h>
+
+static int newthread(void *code_)
+{
+ char *code = code_;
+ lua_State *L = luaL_newstate();
+
+ if(!L) {
+ printf("THREAD FATAL ERROR: could not create lua state\n");
+ return -1;
+ }
+ luaL_openlibs(L);
+
+ if(luaL_loadstring(L, code)) {
+ printf("FATAL THREAD PANIC: (loadstring) %s\n", lua_tolstring(L, -1, NULL));
+ free(code);
+ lua_close(L);
+ return -1;
+ }
+ free(code);
+ if(lua_pcall(L, 0, 0, 0)) {
+ printf("FATAL THREAD PANIC: (pcall) %s\n", lua_tolstring(L, -1, NULL));
+ lua_close(L);
+ return -1;
+ }
+
+ lua_close(L);
+ return 0;
+}
+
+static int thread_new(lua_State *L)
+{
+ THThread *thread = NULL;
+ size_t len = 0;
+ const char *code = luaL_checklstring(L, 1, &len);
+ char *code_dup = malloc(len+1);
+ if(!code_dup)
+ luaL_error(L, "threads: out of memory");
+ memcpy(code_dup, code, len+1);
+
+ thread = THThread_new(newthread, (void*)code_dup);
+ if(!thread)
+ luaL_error(L, "threads: thread new failed");
+ luaTHRD_pushudata(L, thread, "threads.Thread");
+
+ return 1;
+}
+
+static int thread_tostring(lua_State *L)
+{
+ char str[128];
+ THThread *thread = luaTHRD_checkudata(L, 1, "threads.Thread");
+ snprintf(str, 128, "threads.Thread <%lx>", THThread_id(thread));
+ lua_pushstring(L, str);
+ return 1;
+}
+
+static int thread_id(lua_State *L)
+{
+ THThread *thread = luaTHRD_checkudata(L, 1, "threads.Thread");
+ lua_pushinteger(L, THThread_id(thread));
+ return 1;
+}
+
+static int thread_free(lua_State *L)
+{
+ THThread *thread = luaTHRD_checkudata(L, 1, "threads.Thread");
+ THThread_free(thread);
+ return 0;
+}
+
+static int mutex_new(lua_State *L)
+{
+ THMutex *mutex = NULL;
+ if(lua_gettop(L) == 0) {
+ mutex = THMutex_new();
+ }
+ else if(lua_gettop(L) == 1) {
+ long id = luaL_checklong(L, 1);
+ mutex = THMutex_newWithId(id);
+ }
+ else
+ luaL_error(L, "threads: mutex new invalid arguments");
+ if(!mutex)
+ luaL_error(L, "threads: mutex new failed");
+ luaTHRD_pushudata(L, mutex, "threads.Mutex");
+ return 1;
+}
+
+static int mutex_tostring(lua_State *L)
+{
+ char str[128];
+ THMutex *mutex = luaTHRD_checkudata(L, 1, "threads.Mutex");
+ snprintf(str, 128, "threads.Mutex <%lx>", THMutex_id(mutex));
+ lua_pushstring(L, str);
+ return 1;
+}
+
+static int mutex_id(lua_State *L)
+{
+ THMutex *mutex = luaTHRD_checkudata(L, 1, "threads.Mutex");
+ lua_pushinteger(L, THMutex_id(mutex));
+ return 1;
+}
+
+static int mutex_lock(lua_State *L)
+{
+ THMutex *mutex = luaTHRD_checkudata(L, 1, "threads.Mutex");
+ if(THMutex_lock(mutex))
+ luaL_error(L, "threads: mutex lock failed");
+ return 0;
+}
+
+static int mutex_unlock(lua_State *L)
+{
+ THMutex *mutex = luaTHRD_checkudata(L, 1, "threads.Mutex");
+ if(THMutex_unlock(mutex))
+ luaL_error(L, "threads: mutex unlock failed");
+ return 0;
+}
+
+static int mutex_free(lua_State *L)
+{
+ THMutex *mutex = luaTHRD_checkudata(L, 1, "threads.Mutex");
+ THMutex_free(mutex);
+ return 0;
+}
+
+static int condition_new(lua_State *L)
+{
+ THCondition *condition = NULL;
+ if(lua_gettop(L) == 0) {
+ condition = THCondition_new();
+ }
+ else if(lua_gettop(L) == 1) {
+ long id = luaL_checklong(L, 1);
+ condition = THCondition_newWithId(id);
+ }
+ else
+ luaL_error(L, "threads: condition new invalid arguments");
+ if(!condition)
+ luaL_error(L, "threads: condition new failed");
+ luaTHRD_pushudata(L, condition, "threads.Condition");
+ return 1;
+}
+
+static int condition_tostring(lua_State *L)
+{
+ char str[128];
+ THCondition *condition = luaTHRD_checkudata(L, 1, "threads.Condition");
+ snprintf(str, 128, "threads.Condition <%lx>", THCondition_id(condition));
+ lua_pushstring(L, str);
+ return 1;
+}
+
+static int condition_id(lua_State *L)
+{
+ THCondition *condition = luaTHRD_checkudata(L, 1, "threads.Condition");
+ lua_pushinteger(L, THCondition_id(condition));
+ return 1;
+}
+
+static int condition_free(lua_State *L)
+{
+ THCondition *condition = luaTHRD_checkudata(L, 1, "threads.Condition");
+ if(!condition)
+ luaL_error(L, "threads: condition free failed");
+ THCondition_free(condition);
+ return 0;
+}
+
+static int condition_signal(lua_State *L)
+{
+ THCondition *condition = luaTHRD_checkudata(L, 1, "threads.Condition");
+ if(THCondition_signal(condition))
+ luaL_error(L, "threads: condition signal failed");
+ return 0;
+}
+
+static int condition_wait(lua_State *L)
+{
+ THCondition *condition = luaTHRD_checkudata(L, 1, "threads.Condition");
+ THMutex *mutex = luaTHRD_checkudata(L, 2, "threads.Mutex");
+ if(THCondition_wait(condition, mutex))
+ luaL_error(L, "threads: condition wait failed");
+ return 0;
+}
+
+static int semaphore_new(lua_State *L)
+{
+ THSemaphore* semaphore = NULL;
+ if(lua_gettop(L) == 1) {
+ int initValue = luaL_checkinteger(L, 1);
+ semaphore = THSemaphore_new(initValue);
+ }
+ else
+ luaL_error(L, "threads: semaphore new invalid arguments");
+ if (!semaphore)
+ luaL_error(L, "threads: semaphore new failed");
+ luaTHRD_pushudata(L, semaphore, "threads.Semaphore");
+ return 1;
+}
+
+static int semaphore_newfromid(lua_State *L)
+{
+ THSemaphore* semaphore = NULL;
+ if(lua_gettop(L) == 1) {
+ long id = luaL_checklong(L, 1);
+ semaphore = THSemaphore_newWithId(id);
+ }
+ else
+ luaL_error(L, "threads: semaphore new invalid arguments");
+ if(!semaphore)
+ luaL_error(L, "threads: semaphore new failed");
+ luaTHRD_pushudata(L, semaphore, "threads.Semaphore");
+ return 1;
+}
+
+static int semaphore_id(lua_State *L)
+{
+ THSemaphore *semaphore = luaTHRD_checkudata(L, 1, "threads.Semaphore");
+ lua_pushinteger(L, THSemaphore_id(semaphore));
+ return 1;
+}
+
+static int semaphore_signal(lua_State *L)
+{
+ THSemaphore *semaphore = luaTHRD_checkudata(L, 1, "threads.Semaphore");
+ if(THSemaphore_signal(semaphore))
+ luaL_error(L, "threads: semaphore signal failed");
+ return 0;
+}
+
+static int semaphore_wait(lua_State *L)
+{
+ THSemaphore *semaphore = luaTHRD_checkudata(L, 1, "threads.Semaphore");
+ if(THSemaphore_wait(semaphore))
+ luaL_error(L, "threads: semaphore wait failed");
+ return 0;
+}
+
+static int semaphore_trywait(lua_State *L)
+{
+ THSemaphore *semaphore = luaTHRD_checkudata(L, 1, "threads.Semaphore");
+ int wait = THSemaphore_trywait(semaphore);
+ lua_pushinteger(L, wait);
+ return 1;
+}
+
+static void semaphore_free(lua_State *L)
+{
+ THSemaphore *semaphore = luaTHRD_checkudata(L, 1, "threads.Semaphore");
+ if(!semaphore)
+ luaL_error(L, "threads: semaphore free failed");
+ THSemaphore_free(semaphore);
+}
+
+static const struct luaL_Reg thread__ [] = {
+ {"new", thread_new},
+ {"__tostring", thread_tostring},
+ {"id", thread_id},
+ {"free", thread_free},
+ {NULL, NULL}
+};
+
+static const struct luaL_Reg mutex__ [] = {
+ {"new", mutex_new},
+ {"__tostring", mutex_tostring},
+ {"id", mutex_id},
+ {"lock", mutex_lock},
+ {"unlock", mutex_unlock},
+ {"free", mutex_free},
+ {NULL, NULL}
+};
+
+static const struct luaL_Reg condition__ [] = {
+ {"new", condition_new},
+ {"__tostring", condition_tostring},
+ {"id", condition_id},
+ {"signal", condition_signal},
+ {"wait", condition_wait},
+ {"free", condition_free},
+ {NULL, NULL}
+};
+
+static const struct luaL_Reg semaphore__ [] = {
+ {"new", semaphore_new},
+ {"newfromid", semaphore_newfromid},
+ {"id", semaphore_id},
+ {"signal", semaphore_signal},
+ {"wait", semaphore_wait},
+ {"trywait", semaphore_trywait},
+ {"free", semaphore_free},
+ {NULL, NULL}
+};
+
+static void thread_init_pkg(lua_State *L)
+{
+ if(!luaL_newmetatable(L, "threads.Thread"))
+ luaL_error(L, "threads: threads.Thread type already exists");
+ luaL_setfuncs(L, thread__, 0);
+ lua_pushstring(L, "__index");
+ lua_pushvalue(L, -2);
+ lua_rawset(L, -3);
+ lua_pop(L, 1);
+
+ if(!luaL_newmetatable(L, "threads.Mutex"))
+ luaL_error(L, "threads: threads.Mutex type already exists");
+ luaL_setfuncs(L, mutex__, 0);
+ lua_pushstring(L, "__index");
+ lua_pushvalue(L, -2);
+ lua_rawset(L, -3);
+ lua_pop(L, 1);
+
+ if(!luaL_newmetatable(L, "threads.Condition"))
+ luaL_error(L, "threads: threads.Condition type already exists");
+ luaL_setfuncs(L, condition__, 0);
+ lua_pushstring(L, "__index");
+ lua_pushvalue(L, -2);
+ lua_rawset(L, -3);
+ lua_pop(L, 1);
+
+ if(!luaL_newmetatable(L, "threads.Semaphore"))
+ luaL_error(L, "threads: threads.Semaphore type already exists");
+ luaL_setfuncs(L, semaphore__, 0);
+ lua_pushstring(L, "__index");
+ lua_pushvalue(L, -2);
+ lua_rawset(L, -3);
+ lua_pop(L, 1);
+
+ lua_pushstring(L, "Thread");
+ luaTHRD_pushctortable(L, thread_new, "threads.Thread");
+ lua_rawset(L, -3);
+
+ lua_pushstring(L, "Mutex");
+ luaTHRD_pushctortable(L, mutex_new, "threads.Mutex");
+ lua_rawset(L, -3);
+
+ lua_pushstring(L, "Condition");
+ luaTHRD_pushctortable(L, condition_new, "threads.Condition");
+ lua_rawset(L, -3);
+
+ lua_pushstring(L, "Semaphore");
+ luaTHRD_pushctortable(L, semaphore_new, "threads.Semaphore");
+ lua_rawset(L, -3);
+}
diff --git a/fastnn/threads/test/test-low-level.lua b/fastnn/threads/test/test-low-level.lua
new file mode 100644
index 0000000..aea31db
--- /dev/null
+++ b/fastnn/threads/test/test-low-level.lua
@@ -0,0 +1,39 @@
+local t = require 'libthreads'
+
+local m = t.Mutex()
+local c = t.Condition()
+print(string.format('| main thread mutex: 0x%x', m:id()))
+print(string.format('| main thread condition: 0x%x', c:id()))
+
+local code = string.format([[
+ local t = require 'libthreads'
+ local c = t.Condition(%d)
+ print('|| hello from thread')
+ print(string.format('|| thread condition: 0x%%x', c:id()))
+ print('|| doing some stuff in thread...')
+ local x = 0
+ for i=1,10000 do
+ for i=1,10000 do
+ x = x + math.sin(i)
+ end
+ x = x / 10000
+ end
+ print(string.format('|| ...ok (x=%%f)', x))
+ c:signal()
+]], c:id())
+
+print(code)
+
+local thread = t.Thread(code)
+
+
+print('| waiting for thread...')
+m:lock()
+c:wait(m)
+print('| thread finished!')
+
+thread:free()
+m:free()
+c:free()
+
+print('| done')
diff --git a/fastnn/threads/test/test-threads-async.lua b/fastnn/threads/test/test-threads-async.lua
new file mode 100644
index 0000000..68bcd35
--- /dev/null
+++ b/fastnn/threads/test/test-threads-async.lua
@@ -0,0 +1,66 @@
+local threads = require 'threads'
+
+local nthread = 4
+local njob = 100
+
+local pool = threads.Threads(
+ nthread,
+ function(threadid)
+ print('starting a new thread/state number ' .. threadid)
+ end
+)
+
+
+local jobid = 0
+local result -- DO NOT put this in get
+local function get()
+
+ -- fill up the queue as much as we can
+ -- this will not block
+ while jobid < njob and pool:acceptsjob() do
+ jobid = jobid + 1
+
+ pool:addjob(
+ function(jobid)
+ print(string.format('thread ID %d is performing job %d', __threadid, jobid))
+ return string.format("job output from job %d", jobid)
+ end,
+
+ function(jobres)
+ result = jobres
+ end,
+
+ jobid
+ )
+ end
+
+ -- is there still something to do?
+ if pool:hasjob() then
+ pool:dojob() -- yes? do it!
+ if pool:haserror() then -- check for errors
+ pool:synchronize() -- finish everything and throw error
+ end
+ return result
+ end
+
+end
+
+local jobdone = 0
+repeat
+ -- get something asynchronously
+ local res = get()
+
+ -- do something with res (if any)
+ if res then
+ print(res)
+ jobdone = jobdone + 1
+ end
+
+until not res -- until there is nothing remaining
+
+assert(jobid == 100)
+assert(jobdone == 100)
+
+print('PASSED')
+
+pool:terminate()
diff --git a/fastnn/threads/test/test-threads-multiple.lua b/fastnn/threads/test/test-threads-multiple.lua
new file mode 100644
index 0000000..4429696
--- /dev/null
+++ b/fastnn/threads/test/test-threads-multiple.lua
@@ -0,0 +1,15 @@
+local threads = require 'threads'
+
+for i=1,1000 do
+ io.write(string.format('%04d.', tonumber(i)))
+ io.flush()
+ local pool =
+ threads.Threads(
+ 4,
+ function(threadid)
+ require 'torch'
+ end
+ )
+end
+print()
+print('PASSED')
diff --git a/fastnn/threads/test/test-threads-shared.lua b/fastnn/threads/test/test-threads-shared.lua
new file mode 100644
index 0000000..3d63851
--- /dev/null
+++ b/fastnn/threads/test/test-threads-shared.lua
@@ -0,0 +1,111 @@
+require 'torch'
+
+local threads = require 'threads'
+local status, tds = pcall(require, 'tds')
+tds = status and tds or nil
+
+local nthread = 4
+local njob = 10
+local msg = "hello from a satellite thread"
+
+threads.Threads.serialization('threads.sharedserialize')
+
+local x = {}
+local xh = tds and tds.hash() or {}
+local xs = {}
+local z = tds and tds.hash() or {}
+local D = 10
+local K = tds and 100000 or 100 -- good luck in non-shared (30M)
+for i=1,njob do
+ x[i] = torch.ones(D)
+ xh[i] = torch.ones(D)
+ xs[i] = torch.FloatStorage(D):fill(1)
+ for j=1,K do
+ z[(i-1)*K+j] = "blah" .. i .. j
+ end
+end
+collectgarbage()
+collectgarbage()
+
+print('GO')
+
+local pool = threads.Threads(
+ nthread,
+ function(threadIdx)
+ pcall(require, 'tds')
+ print('starting a new thread/state number:', threadIdx)
+ gmsg = msg -- we copy here an upvalue of the main thread
+ end
+)
+
+local jobdone = 0
+for i=1,njob do
+ pool:addjob(
+ function()
+ assert(x[i]:sum() == D)
+ assert(xh[i]:sum() == D)
+ assert(torch.FloatTensor(xs[i]):sum() == D)
+ for j=1,K do
+ assert(z[(i-1)*K+j] == "blah" .. i .. j)
+ end
+ x[i]:add(1)
+ xh[i]:add(1)
+ torch.FloatTensor(xs[i]):add(1)
+ print(string.format('%s -- thread ID is %x', gmsg, __threadid))
+ collectgarbage()
+ collectgarbage()
+ return __threadid
+ end,
+
+ function(id)
+ print(string.format("task %d finished (ran on thread ID %x)", i, id))
+ jobdone = jobdone + 1
+ end
+ )
+end
+
+for i=1,njob do
+ pool:addjob(
+ function()
+ collectgarbage()
+ collectgarbage()
+ end
+ )
+end
+
+pool:synchronize()
+
+print(string.format('%d jobs done', jobdone))
+
+pool:terminate()
+
+-- did we do the job in shared mode?
+for i=1,njob do
+ assert(x[i]:sum() == 2*D)
+ assert(xh[i]:sum() == 2*D)
+ assert(torch.FloatTensor(xs[i]):sum() == 2*D)
+end
+
+-- serialize and zero x
+local str = torch.serialize(x)
+local strh = torch.serialize(xh)
+local strs = torch.serialize(xs)
+for i=1,njob do
+ x[i]:zero()
+ xh[i]:zero()
+ xs[i]:fill(0)
+end
+
+-- dude, check that unserialized x does not point on x
+local y = torch.deserialize(str)
+local yh = torch.deserialize(strh)
+local ys = torch.deserialize(strs)
+for i=1,njob do
+ assert(y[i]:sum() == 2*D)
+ assert(yh[i]:sum() == 2*D)
+ assert(torch.FloatTensor(ys[i]):sum() == 2*D)
+end
+
+pool:terminate()
+
+print('PASSED')
diff --git a/fastnn/threads/test/test-threads.lua b/fastnn/threads/test/test-threads.lua
new file mode 100644
index 0000000..e49a381
--- /dev/null
+++ b/fastnn/threads/test/test-threads.lua
@@ -0,0 +1,20 @@
+local clib = require 'libthreads'
+
+
+nthread = 1
+
+str='ad;alkfakd;af'
+code = [[ function func(str) print(str) end; print(str);]]
+print(code)
+
+--thread = clib.Thread(code)
+
+
+--thread:free()
+
+
+require 'threads'
+
+tt = threads.Thread(code)
+
+
diff --git a/fastnn/threads/threads-scm-1.rockspec b/fastnn/threads/threads-scm-1.rockspec
new file mode 100644
index 0000000..02535d4
--- /dev/null
+++ b/fastnn/threads/threads-scm-1.rockspec
@@ -0,0 +1,36 @@
+package = "threads"
+version = "scm-1"
+source = {
+ url = "https://github.com/uphantom/nerv-threads.git"
+}
+description = {
+ summary = "Thread control for Nerv fastnn",
+ detailed = [[
+ ]],
+ homepage = "https://github.com/uphantom/nerv-threads",
+ license = "BSD"
+}
+dependencies = {
+ "nerv >= scm-1",
+ "lua >= 5.1"
+}
+build = {
+ type = "make",
+ build_variables = {
+ CFLAGS="$(CFLAGS)",
+ LIBFLAG="$(LIBFLAG)",
+ LUA_LIBDIR="$(LUA_LIBDIR)",
+ LUA_BINDIR="$(LUA_BINDIR)",
+ LUA_INCDIR="$(LUA_INCDIR)",
+ INST_PREFIX="$(PREFIX)",
+ LUA="$(LUA)",
+ },
+ install_variables = {
+ LUA_BINDIR="$(LUA_BINDIR)",
+ INST_PREFIX="$(PREFIX)",
+ INST_BINDIR="$(BINDIR)",
+ INST_LIBDIR="$(LIBDIR)",
+ INST_LUADIR="$(LUADIR)",
+ INST_CONFDIR="$(CONFDIR)",
+ },
+}
diff --git a/fastnn/threads/threads.lua b/fastnn/threads/threads.lua
new file mode 100644
index 0000000..2db7d51
--- /dev/null
+++ b/fastnn/threads/threads.lua
@@ -0,0 +1,14 @@
+local threads = {}
+
+local C = require 'libthreads'
+
+threads.Thread = C.Thread
+threads.Mutex = C.Mutex
+threads.Condition = C.Condition
+threads.Semaphore = C.Semaphore
+--threads.Threads = require 'threads.threads'
+
+-- only for backward compatibility (boo)
+-- setmetatable(threads, getmetatable(threads.Threads))
+
+-- return threads