mirror of https://github.com/oxen-io/lokinet
refactor threadpool
This commit is contained in:
Jeff Becker
parent
d31391d856
commit
a7703b0dbc
2
Makefile
2
Makefile
|
|
@ -76,7 +76,7 @@ $(REPO)/test/%.c.bin: $(REPO)/test/%.c
|
|||
$(CC) $(REQUIRED_CFLAGS) $< -o $<.bin $(TEST_LDFLAGS)
|
||||
mv $<.bin $<.test
|
||||
$<.test
|
||||
valgrind --tool=callgrind $<.test
|
||||
#valgrind --tool=callgrind $<.test
|
||||
|
||||
$(EXE): $(DAEMON_OBJ) $(STATIC_LIB)
|
||||
$(CXX) $(DAEMON_OBJ) $(STATIC_LIB) $(LIB_LDFLAGS) -o $(EXE)
|
||||
|
|
|
|||
|
|
@ -12,8 +12,8 @@ struct llarp_async_dh;
|
|||
|
||||
struct llarp_async_dh *llarp_async_dh_new(llarp_seckey_t ourkey,
|
||||
struct llarp_crypto *crypto,
|
||||
struct llarp_ev_loop *ev,
|
||||
struct llarp_threadpool *tp);
|
||||
struct llarp_threadpool *handler,
|
||||
struct llarp_threadpool *worker);
|
||||
void llarp_async_dh_free(struct llarp_async_dh **dh);
|
||||
|
||||
struct llarp_dh_result;
|
||||
|
|
@ -47,8 +47,8 @@ struct llarp_cipher_result {
|
|||
|
||||
struct llarp_async_cipher *llarp_async_cipher_new(llarp_sharedkey_t key,
|
||||
struct llarp_crypto *crypto,
|
||||
struct llarp_ev_loop *ev,
|
||||
struct llarp_threadpool *tp);
|
||||
struct llarp_threadpool *result,
|
||||
struct llarp_threadpool *worker);
|
||||
|
||||
void llarp_async_cipher_free(struct llarp_async_cipher **c);
|
||||
|
||||
|
|
|
|||
|
|
@ -33,35 +33,6 @@ int llarp_ev_add_udp(struct llarp_ev_loop *ev,
|
|||
int llarp_ev_udp_sendto(struct llarp_udp_io * udp, const struct sockaddr * to, const void * data, size_t sz);
|
||||
|
||||
int llarp_ev_close_udp(struct llarp_udp_io * udp);
|
||||
|
||||
struct llarp_ev_async_call;
|
||||
|
||||
typedef void (*llarp_ev_work_func)(struct llarp_ev_async_call *);
|
||||
|
||||
struct llarp_ev_caller;
|
||||
|
||||
struct llarp_ev_async_call {
|
||||
/** the loop this job belongs to */
|
||||
const struct llarp_ev_loop *loop;
|
||||
/** private implementation */
|
||||
const struct llarp_ev_caller *parent;
|
||||
/** user data */
|
||||
const void *user;
|
||||
/**
|
||||
work is called async when ready in the event loop thread
|
||||
must not free from inside this call as it is done elsewhere
|
||||
*/
|
||||
const llarp_ev_work_func work;
|
||||
};
|
||||
|
||||
struct llarp_ev_caller *llarp_ev_prepare_async(struct llarp_ev_loop *ev,
|
||||
llarp_ev_work_func func);
|
||||
|
||||
bool llarp_ev_call_async(struct llarp_ev_caller *c, void *user);
|
||||
|
||||
bool larp_ev_call_many_async(struct llarp_ev_caller *c, void **users, size_t n);
|
||||
|
||||
void llarp_ev_caller_stop(struct llarp_ev_caller *c);
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
|
|
|
|||
|
|
@ -14,12 +14,6 @@ typedef void (*llarp_thread_work_func)(void *);
|
|||
|
||||
/** job to be done in worker thread */
|
||||
struct llarp_thread_job {
|
||||
/**
|
||||
called async after work is executed
|
||||
*/
|
||||
struct llarp_ev_caller *caller;
|
||||
void *data;
|
||||
|
||||
/** user data to pass to work function */
|
||||
void *user;
|
||||
/** called in threadpool worker thread */
|
||||
|
|
@ -30,7 +24,10 @@ void llarp_threadpool_queue_job(struct llarp_threadpool *tp,
|
|||
struct llarp_thread_job j);
|
||||
|
||||
void llarp_threadpool_start(struct llarp_threadpool *tp);
|
||||
void llarp_threadpool_stop(struct llarp_threadpool * tp);
|
||||
void llarp_threadpool_join(struct llarp_threadpool *tp);
|
||||
|
||||
void llarp_threadpool_wait(struct llarp_threadpool *tp);
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,29 +5,34 @@
|
|||
struct llarp_async_dh {
|
||||
llarp_dh_func client;
|
||||
llarp_dh_func server;
|
||||
struct llarp_threadpool *tp;
|
||||
struct llarp_ev_caller *caller;
|
||||
struct llarp_threadpool *worker;
|
||||
struct llarp_threadpool *result;
|
||||
llarp_seckey_t ourkey;
|
||||
};
|
||||
|
||||
struct llarp_dh_internal {
|
||||
llarp_pubkey_t theirkey;
|
||||
uint8_t *ourkey;
|
||||
llarp_tunnel_nounce_t nounce;
|
||||
struct llarp_dh_result result;
|
||||
llarp_dh_func func;
|
||||
struct llarp_async_dh * parent;
|
||||
};
|
||||
|
||||
|
||||
static void llarp_crypto_dh_result(void *call) {
|
||||
struct llarp_dh_internal *impl = (struct llarp_dh_internal *)call;
|
||||
impl->result.hook(&impl->result);
|
||||
llarp_g_mem.free(impl);
|
||||
}
|
||||
|
||||
|
||||
static void llarp_crypto_dh_work(void *user) {
|
||||
struct llarp_dh_internal *impl = (struct llarp_dh_internal *)user;
|
||||
impl->func(&impl->result.sharedkey, impl->theirkey, impl->nounce,
|
||||
impl->ourkey);
|
||||
}
|
||||
|
||||
static void llarp_crypto_dh_result(struct llarp_ev_async_call *call) {
|
||||
struct llarp_dh_internal *impl = (struct llarp_dh_internal *)call->user;
|
||||
impl->result.hook(&impl->result);
|
||||
llarp_g_mem.free(impl);
|
||||
impl->parent->ourkey);
|
||||
struct llarp_thread_job job = { .user = impl,
|
||||
.work = &llarp_crypto_dh_result };
|
||||
llarp_threadpool_queue_job(impl->parent->result, job);
|
||||
}
|
||||
|
||||
static void llarp_async_dh_exec(struct llarp_async_dh *dh, llarp_dh_func func,
|
||||
|
|
@ -36,17 +41,15 @@ static void llarp_async_dh_exec(struct llarp_async_dh *dh, llarp_dh_func func,
|
|||
llarp_dh_complete_hook result, void *user) {
|
||||
struct llarp_dh_internal *impl =
|
||||
llarp_g_mem.alloc(sizeof(struct llarp_dh_internal), 32);
|
||||
struct llarp_thread_job job = {.caller = dh->caller,
|
||||
.data = impl,
|
||||
.user = impl,
|
||||
struct llarp_thread_job job = {.user = impl,
|
||||
.work = &llarp_crypto_dh_work};
|
||||
memcpy(impl->theirkey, theirkey, sizeof(llarp_pubkey_t));
|
||||
memcpy(impl->nounce, nounce, sizeof(llarp_tunnel_nounce_t));
|
||||
impl->ourkey = dh->ourkey;
|
||||
impl->parent = dh;
|
||||
impl->result.user = user;
|
||||
impl->result.hook = result;
|
||||
impl->func = func;
|
||||
llarp_threadpool_queue_job(dh->tp, job);
|
||||
llarp_threadpool_queue_job(dh->worker, job);
|
||||
}
|
||||
|
||||
void llarp_async_client_dh(struct llarp_async_dh *dh, llarp_pubkey_t theirkey,
|
||||
|
|
@ -63,43 +66,42 @@ void llarp_async_server_dh(struct llarp_async_dh *dh, llarp_pubkey_t theirkey,
|
|||
|
||||
struct llarp_async_dh *llarp_async_dh_new(llarp_seckey_t ourkey,
|
||||
struct llarp_crypto *crypto,
|
||||
struct llarp_ev_loop *ev,
|
||||
struct llarp_threadpool *tp) {
|
||||
struct llarp_threadpool *result,
|
||||
struct llarp_threadpool *worker) {
|
||||
struct llarp_async_dh *dh =
|
||||
llarp_g_mem.alloc(sizeof(struct llarp_async_dh), 16);
|
||||
dh->client = crypto->dh_client;
|
||||
dh->server = crypto->dh_server;
|
||||
memcpy(dh->ourkey, ourkey, sizeof(llarp_seckey_t));
|
||||
dh->tp = tp;
|
||||
dh->caller = llarp_ev_prepare_async(ev, &llarp_crypto_dh_result);
|
||||
dh->result = result;
|
||||
dh->worker = worker;
|
||||
return dh;
|
||||
}
|
||||
|
||||
void llarp_async_dh_free(struct llarp_async_dh **dh) {
|
||||
if (*dh) {
|
||||
llarp_ev_caller_stop((*dh)->caller);
|
||||
llarp_g_mem.free(*dh);
|
||||
*dh = NULL;
|
||||
}
|
||||
}
|
||||
|
||||
struct llarp_async_cipher_internal {
|
||||
uint8_t *key;
|
||||
llarp_nounce_t nounce;
|
||||
struct llarp_cipher_result result;
|
||||
struct llarp_async_cipher * parent;
|
||||
llarp_sym_cipher_func func;
|
||||
};
|
||||
|
||||
struct llarp_async_cipher {
|
||||
llarp_sharedkey_t key;
|
||||
struct llarp_threadpool *tp;
|
||||
struct llarp_ev_caller *caller;
|
||||
struct llarp_threadpool *worker;
|
||||
struct llarp_threadpool *result;
|
||||
llarp_sym_cipher_func func;
|
||||
};
|
||||
|
||||
static void llarp_crypto_cipher_result(struct llarp_ev_async_call *job) {
|
||||
static void llarp_crypto_cipher_result(void *user) {
|
||||
struct llarp_async_cipher_internal *impl =
|
||||
(struct llarp_async_cipher_internal *)job->user;
|
||||
(struct llarp_async_cipher_internal *)user;
|
||||
impl->result.hook(&impl->result);
|
||||
llarp_g_mem.free(impl);
|
||||
}
|
||||
|
|
@ -107,7 +109,12 @@ static void llarp_crypto_cipher_result(struct llarp_ev_async_call *job) {
|
|||
static void llarp_crypto_cipher_work(void *work) {
|
||||
struct llarp_async_cipher_internal *impl =
|
||||
(struct llarp_async_cipher_internal *)work;
|
||||
impl->func(impl->result.buff, impl->key, impl->nounce);
|
||||
impl->func(impl->result.buff, impl->parent->key, impl->nounce);
|
||||
struct llarp_thread_job job = {
|
||||
.user = impl,
|
||||
.work = &llarp_crypto_cipher_result
|
||||
};
|
||||
llarp_threadpool_queue_job(impl->parent->result, job);
|
||||
}
|
||||
|
||||
void llarp_async_cipher_queue_op(struct llarp_async_cipher *c,
|
||||
|
|
@ -115,36 +122,33 @@ void llarp_async_cipher_queue_op(struct llarp_async_cipher *c,
|
|||
llarp_cipher_complete_hook h, void *user) {
|
||||
struct llarp_async_cipher_internal *impl =
|
||||
llarp_g_mem.alloc(sizeof(struct llarp_async_cipher_internal), 16);
|
||||
impl->key = c->key;
|
||||
impl->parent = c;
|
||||
memcpy(impl->nounce, n, sizeof(llarp_nounce_t));
|
||||
impl->result.user = user;
|
||||
impl->result.buff.base = buff->base;
|
||||
impl->result.buff.sz = buff->sz;
|
||||
impl->result.hook = h;
|
||||
impl->func = c->func;
|
||||
struct llarp_thread_job job = {.caller = c->caller,
|
||||
.data = impl,
|
||||
.user = impl,
|
||||
struct llarp_thread_job job = {.user = impl,
|
||||
.work = &llarp_crypto_cipher_work};
|
||||
llarp_threadpool_queue_job(c->tp, job);
|
||||
llarp_threadpool_queue_job(c->worker, job);
|
||||
}
|
||||
|
||||
struct llarp_async_cipher *llarp_async_cipher_new(llarp_sharedkey_t key,
|
||||
struct llarp_crypto *crypto,
|
||||
struct llarp_ev_loop *ev,
|
||||
struct llarp_threadpool *tp) {
|
||||
struct llarp_threadpool *result,
|
||||
struct llarp_threadpool *worker) {
|
||||
struct llarp_async_cipher *cipher =
|
||||
llarp_g_mem.alloc(sizeof(struct llarp_async_cipher), 16);
|
||||
cipher->func = crypto->xchacha20;
|
||||
cipher->tp = tp;
|
||||
cipher->caller = llarp_ev_prepare_async(ev, &llarp_crypto_cipher_result);
|
||||
cipher->result = result;
|
||||
cipher->worker = worker;
|
||||
memcpy(cipher->key, key, sizeof(llarp_sharedkey_t));
|
||||
return cipher;
|
||||
}
|
||||
|
||||
void llarp_async_cipher_free(struct llarp_async_cipher **c) {
|
||||
if (*c) {
|
||||
llarp_ev_caller_stop((*c)->caller);
|
||||
llarp_g_mem.free(*c);
|
||||
*c = NULL;
|
||||
}
|
||||
|
|
|
|||
66
llarp/ev.cpp
66
llarp/ev.cpp
|
|
@ -8,56 +8,6 @@
|
|||
#include "ev_kqueue.hpp"
|
||||
#endif
|
||||
|
||||
#include <mutex>
|
||||
#include <queue>
|
||||
|
||||
struct llarp_ev_caller {
|
||||
static void *operator new(size_t sz) {
|
||||
return llarp::Alloc<llarp_ev_caller>();
|
||||
}
|
||||
|
||||
static void operator delete(void *ptr) { llarp_g_mem.free(ptr); }
|
||||
|
||||
llarp_ev_caller(llarp_ev_loop *ev, llarp_ev_work_func func)
|
||||
: loop(ev), work(func) {
|
||||
}
|
||||
|
||||
~llarp_ev_caller() {}
|
||||
|
||||
bool appendCall(void *user) {
|
||||
std::unique_lock<std::mutex> lock(access);
|
||||
pending.emplace_back(
|
||||
std::move(llarp_ev_async_call{loop, this, user, this->work}));
|
||||
return true;
|
||||
}
|
||||
|
||||
bool appendManyCalls(void **users, size_t n) {
|
||||
std::unique_lock<std::mutex> lock(access);
|
||||
while (n--) {
|
||||
pending.emplace_back(
|
||||
std::move(llarp_ev_async_call{loop, this, *users, this->work}));
|
||||
users++;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
void Call() {
|
||||
std::unique_lock<std::mutex> lock(access);
|
||||
auto sz = pending.size();
|
||||
while (sz > 0) {
|
||||
auto &front = pending.front();
|
||||
front.work(&front);
|
||||
pending.pop_front();
|
||||
--sz;
|
||||
}
|
||||
}
|
||||
|
||||
std::mutex access;
|
||||
struct llarp_ev_loop *loop;
|
||||
std::deque<llarp_ev_async_call> pending;
|
||||
llarp_ev_work_func work;
|
||||
};
|
||||
|
||||
extern "C" {
|
||||
void llarp_ev_loop_alloc(struct llarp_ev_loop **ev) {
|
||||
#ifdef __linux__
|
||||
|
|
@ -90,21 +40,5 @@ int llarp_ev_close_udp_listener(struct llarp_udp_listener *listener) {
|
|||
|
||||
void llarp_ev_loop_stop(struct llarp_ev_loop *loop) { loop->stop(); }
|
||||
|
||||
struct llarp_ev_caller *llarp_ev_prepare_async(struct llarp_ev_loop *loop,
|
||||
llarp_ev_work_func work) {
|
||||
return new llarp_ev_caller(loop, work);
|
||||
}
|
||||
|
||||
bool llarp_ev_call_async(struct llarp_ev_caller *caller, void *user) {
|
||||
return false;
|
||||
}
|
||||
|
||||
bool llarp_ev_call_many_async(struct llarp_ev_caller *caller, void **users,
|
||||
size_t n) {
|
||||
return false;
|
||||
}
|
||||
|
||||
void llarp_ev_caller_stop(struct llarp_ev_caller *caller) {
|
||||
delete caller;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,5 +1,4 @@
|
|||
#include "threadpool.hpp"
|
||||
#include <iostream>
|
||||
|
||||
namespace llarp {
|
||||
namespace thread {
|
||||
|
|
@ -19,23 +18,22 @@ Pool::Pool(size_t workers) {
|
|||
}
|
||||
// do work
|
||||
job.work(job.user);
|
||||
// inform result if needed
|
||||
if (job.caller) {
|
||||
if (!llarp_ev_call_async(job.caller, job.data)) {
|
||||
std::cerr << "failed to queue result in thread worker" << std::endl;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
void Pool::Join() {
|
||||
void Pool::Stop()
|
||||
{
|
||||
{
|
||||
lock_t lock(queue_mutex);
|
||||
stop = true;
|
||||
}
|
||||
condition.notify_all();
|
||||
done.notify_all();
|
||||
}
|
||||
|
||||
void Pool::Join() {
|
||||
for (auto &t : threads) t.join();
|
||||
}
|
||||
|
||||
|
|
@ -74,6 +72,20 @@ void llarp_threadpool_join(struct llarp_threadpool *pool) { pool->impl.Join(); }
|
|||
void llarp_threadpool_start(struct llarp_threadpool *pool) { /** no op */
|
||||
}
|
||||
|
||||
void llarp_threadpool_stop(struct llarp_threadpool *pool)
|
||||
{
|
||||
pool->impl.Stop();
|
||||
}
|
||||
|
||||
void llarp_threadpool_wait(struct llarp_threadpool *pool)
|
||||
{
|
||||
std::mutex mtx;
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(mtx);
|
||||
pool->impl.done.wait(lock);
|
||||
}
|
||||
}
|
||||
|
||||
void llarp_threadpool_queue_job(struct llarp_threadpool *pool,
|
||||
struct llarp_thread_job job) {
|
||||
pool->impl.QueueJob(job);
|
||||
|
|
|
|||
|
|
@ -17,11 +17,13 @@ struct Pool {
|
|||
Pool(size_t sz);
|
||||
void QueueJob(const llarp_thread_job& job);
|
||||
void Join();
|
||||
void Stop();
|
||||
std::vector<std::thread> threads;
|
||||
std::deque<llarp_thread_job> jobs;
|
||||
|
||||
mtx_t queue_mutex;
|
||||
std::condition_variable condition;
|
||||
std::condition_variable done;
|
||||
bool stop;
|
||||
};
|
||||
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ struct bench_main {
|
|||
size_t completed;
|
||||
size_t num;
|
||||
size_t jobs;
|
||||
struct llarp_ev_loop *ev;
|
||||
struct llarp_threadpool * pool;
|
||||
struct llarp_async_cipher *cipher;
|
||||
struct llarp_crypto crypto;
|
||||
};
|
||||
|
|
@ -20,7 +20,8 @@ static void handle_cipher_complete(struct llarp_cipher_result *res) {
|
|||
if (m->completed % 10000 == 0)
|
||||
printf("completed %ld and %ld left\n", m->completed, left);
|
||||
if (m->completed == m->num) {
|
||||
llarp_ev_loop_stop(m->ev);
|
||||
llarp_threadpool_stop(m->pool);
|
||||
printf("done\n");
|
||||
} else if (m->completed % sz == 0) {
|
||||
llarp_nounce_t nounce;
|
||||
while (sz--) {
|
||||
|
|
@ -37,9 +38,9 @@ int main(int argc, char *argv[]) {
|
|||
|
||||
llarp_mem_stdlib();
|
||||
llarp_crypto_libsodium_init(&b_main.crypto);
|
||||
llarp_ev_loop_alloc(&b_main.ev);
|
||||
|
||||
tp = llarp_init_threadpool(2);
|
||||
b_main.pool = llarp_init_threadpool(1);
|
||||
llarp_threadpool_start(b_main.pool);
|
||||
tp = llarp_init_threadpool(8);
|
||||
|
||||
b_main.num = 500000;
|
||||
b_main.jobs = 10;
|
||||
|
|
@ -47,7 +48,7 @@ int main(int argc, char *argv[]) {
|
|||
llarp_sharedkey_t key;
|
||||
b_main.crypto.randbytes(key, sizeof(llarp_sharedkey_t));
|
||||
|
||||
b_main.cipher = llarp_async_cipher_new(key, &b_main.crypto, b_main.ev, tp);
|
||||
b_main.cipher = llarp_async_cipher_new(key, &b_main.crypto, b_main.pool, tp);
|
||||
llarp_threadpool_start(tp);
|
||||
|
||||
llarp_nounce_t nounce;
|
||||
|
|
@ -68,13 +69,14 @@ int main(int argc, char *argv[]) {
|
|||
llarp_async_cipher_queue_op(b_main.cipher, msg, nounce,
|
||||
handle_cipher_complete, &b_main);
|
||||
}
|
||||
llarp_ev_loop_run(b_main.ev);
|
||||
|
||||
|
||||
llarp_threadpool_wait(b_main.pool);
|
||||
llarp_threadpool_join(b_main.pool);
|
||||
llarp_threadpool_stop(tp);
|
||||
llarp_threadpool_join(tp);
|
||||
llarp_async_cipher_free(&b_main.cipher);
|
||||
|
||||
llarp_ev_loop_free(&b_main.ev);
|
||||
llarp_free_threadpool(&tp);
|
||||
llarp_free_threadpool(&b_main.pool);
|
||||
llarp_async_cipher_free(&b_main.cipher);
|
||||
printf("did %ld of %ld work\n", b_main.completed, b_main.num);
|
||||
return 0;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@
|
|||
struct dh_bench_main {
|
||||
size_t completed;
|
||||
size_t num;
|
||||
struct llarp_ev_loop *ev;
|
||||
struct llarp_threadpool *result;
|
||||
struct llarp_async_dh *dh;
|
||||
};
|
||||
|
||||
|
|
@ -16,7 +16,8 @@ static void handle_dh_complete(struct llarp_dh_result *res) {
|
|||
m->completed++;
|
||||
if (m->completed % 10000 == 0) printf("completed %ld\n", m->completed);
|
||||
if (m->completed == m->num) {
|
||||
llarp_ev_loop_stop(m->ev);
|
||||
printf("done\n");
|
||||
llarp_threadpool_stop(m->result);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -27,11 +28,12 @@ int main(int argc, char *argv[]) {
|
|||
|
||||
llarp_mem_stdlib();
|
||||
llarp_crypto_libsodium_init(&crypto);
|
||||
llarp_ev_loop_alloc(&dh_main.ev);
|
||||
|
||||
tp = llarp_init_threadpool(2);
|
||||
|
||||
dh_main.num = 10000;
|
||||
tp = llarp_init_threadpool(8);
|
||||
dh_main.result = llarp_init_threadpool(1);
|
||||
llarp_threadpool_start(dh_main.result);
|
||||
|
||||
dh_main.num = 500000;
|
||||
dh_main.completed = 0;
|
||||
llarp_seckey_t ourkey;
|
||||
llarp_seckey_t theirkey;
|
||||
|
|
@ -39,7 +41,7 @@ int main(int argc, char *argv[]) {
|
|||
crypto.keygen(&ourkey);
|
||||
crypto.keygen(&theirkey);
|
||||
|
||||
dh_main.dh = llarp_async_dh_new(ourkey, &crypto, dh_main.ev, tp);
|
||||
dh_main.dh = llarp_async_dh_new(ourkey, &crypto, dh_main.result, tp);
|
||||
llarp_threadpool_start(tp);
|
||||
|
||||
llarp_tunnel_nounce_t nounce;
|
||||
|
|
@ -59,13 +61,15 @@ int main(int argc, char *argv[]) {
|
|||
&dh_main);
|
||||
}
|
||||
printf("started %ld dh jobs\n", dh_main.num);
|
||||
llarp_ev_loop_run(dh_main.ev);
|
||||
|
||||
llarp_threadpool_wait(dh_main.result);
|
||||
llarp_threadpool_join(dh_main.result);
|
||||
llarp_threadpool_stop(tp);
|
||||
llarp_threadpool_join(tp);
|
||||
llarp_async_dh_free(&dh_main.dh);
|
||||
|
||||
llarp_ev_loop_free(&dh_main.ev);
|
||||
|
||||
llarp_free_threadpool(&tp);
|
||||
llarp_free_threadpool(&dh_main.result);
|
||||
|
||||
llarp_async_dh_free(&dh_main.dh);
|
||||
printf("did %ld of %ld work\n", dh_main.completed, dh_main.num);
|
||||
return 0;
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue