eNifWorkers/c_src/NifWorkers/eNifWorkers.cc

#include<stdio.h>
#include<stdlib.h>
#include<stdbool.h>
#include<atomic>
#include<vector>
#include<erl_nif.h>

#include "eNifWorkers.h"
#include "blockingconcurrentqueue.h"

static void *UpdateMapFun(void *obj);
static void *FindPathFun(void *obj);

struct NifTraits : public moodycamel::ConcurrentQueueDefaultTraits {
    static const size_t BLOCK_SIZE = 16;
    static const size_t EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD = 16;
    static const size_t EXPLICIT_INITIAL_INDEX_SIZE = 8;
    static const size_t IMPLICIT_INITIAL_INDEX_SIZE = 8;
    static const size_t INITIAL_IMPLICIT_PRODUCER_HASH_SIZE = 16;
    static const std::uint32_t EXPLICIT_CONSUMER_CONSUMPTION_QUOTA_BEFORE_ROTATE = 256;

    static inline void *malloc(std::size_t size) { return enif_alloc(size); }
    static inline void free(void *ptr) { enif_free(ptr); }
};

typedef struct UpdateMapData_r {
    bool IsBarrier;            
} UpdateMapData;

typedef struct FindPathData_r {
    bool IsBarrier;                
} FindPathData;


using LfqUpdateMap = moodycamel::BlockingConcurrentQueue<UpdateMapData, NifTraits> *;
using LfqFindPath = moodycamel::BlockingConcurrentQueue<FindPathData, NifTraits> *;

LfqUpdateMap QUpdateMap;
LfqFindPath QFindPath;

const size_t BulkDelCnt = 200;


typedef struct Node_r {
    bool IsBarrier;                 // 是否障碍
} Node;

int nifLoad(ErlNifEnv* env, void** priv_data, ERL_NIF_TERM load_info) {
	enif_fprintf(stdout, "IMY*************nifload00000\n");
	NIF_ATOMS(NIF_ATOM_INIT)

	enif_fprintf(stdout, "IMY*************nifload00001\n");
	*priv_data = NULL;
	QUpdateMap = new moodycamel::BlockingConcurrentQueue<UpdateMapData, NifTraits>;
	QFindPath = new moodycamel::BlockingConcurrentQueue<FindPathData, NifTraits>;
    ErlNifTid Tid;
	if (0 != enif_thread_create((char *)"ThreadUpdateMap", &Tid, UpdateMapFun, NULL, NULL)) {
        return -1;
    }

	if (0 != enif_thread_create((char *)"FindPathFun", &Tid, FindPathFun, NULL, NULL)) {
        return -1;
    }

    if (0 != enif_thread_create((char *)"FindPathFun", &Tid, FindPathFun, NULL, NULL)) {
        return -1;
    }

    if (0 != enif_thread_create((char *)"FindPathFun", &Tid, FindPathFun, NULL, NULL)) {
        return -1;
    }

	return 0;
}

int nifUpgrade(ErlNifEnv* env, void** priv_data, void** old_priv_data, ERL_NIF_TERM load_info) {
	*priv_data = *old_priv_data;
	enif_fprintf(stdout, "IMY*************nifUpgrade %p %T\n", old_priv_data, load_info);
	return 0;
}

void nifUnload(ErlNifEnv* env, void* priv_data) {
	enif_fprintf(stdout, "IMY*************nifUnload0000 \n");
	return;
}

ERL_NIF_TERM nifInitMap(ErlNifEnv *env, int, const ERL_NIF_TERM argv[]) {
	return atom_ok;
}

ERL_NIF_TERM nifUpdateMap(ErlNifEnv *env, int, const ERL_NIF_TERM argv[]) {
    UpdateMapData InsertData = {true};
  
    if (QUpdateMap->enqueue(InsertData)) {
        return atom_true;
    } else {
        return atom_false;
    }
}

ERL_NIF_TERM nifFindPath(ErlNifEnv *env, int, const ERL_NIF_TERM argv[]) {
    FindPathData InsertData = {true};
  
    if (QFindPath->enqueue(InsertData)) {
        return atom_true;
    } else {
        return atom_false;
    }
}

static ErlNifFunc nifFuns[] = {
	{"initMap", 1, nifInitMap},
	{"updateMap", 3, nifUpdateMap},
	{"findPath", 5, nifFindPath}
};

ERL_NIF_INIT(eNifWorkers, nifFuns, nifLoad, NULL, nifUpgrade, nifUnload)

static void *UpdateMapFun(void *obj) {

    return (void *) 0; // 线程处理结束，退出
}

static void *FindPathFun(void *obj) {
    FindPathData Data;
    ErlNifTid  Tid = enif_thread_self();
    while (true)
    {
        QFindPath->wait_dequeue(Data);
        enif_fprintf(stdout, "IMY*************listenerResClose5 %T %d\n", Tid, Data.IsBarrier);
    }
    
    return (void *) 0; // 线程处理结束，退出
}

void eLfqFree(ErlNifEnv *, void *obj) {
        std::vector <FindPathData> DataList(BulkDelCnt);
        size_t OutSize;
        do{
            OutSize = QFindPath->try_dequeue_bulk(DataList.begin(), DataList.size());
        }while(OutSize >= BulkDelCnt);
   
}