greg/wterl
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Replace all enif implementations of mutexes and conditions with their POSIX pthread equivalent on the theory that we don't want to be bumping heads with the Erlang runtime.

Browse source
This commit is contained in:
Gregory Burd 2013-08-02 14:20:04 -04:00
parent e9b1a9ea0b
commit ef3bc102f2
1 changed files with 73 additions and 72 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

145
c_src/wterl.c
View file

@ -25,6 +25,7 @@
#include <stdarg.h> #include <stdarg.h>
#include <inttypes.h> #include <inttypes.h>
#include <errno.h> #include <errno.h>
#include <pthread.h>
#include "wiredtiger.h" #include "wiredtiger.h"
@ -57,7 +58,7 @@ typedef struct wterl_conn {
WT_CONNECTION *conn; WT_CONNECTION *conn;
const char *session_config; const char *session_config;
STAILQ_HEAD(ctxs, wterl_ctx) cache; STAILQ_HEAD(ctxs, wterl_ctx) cache;
ErlNifMutex *cache_mutex; pthread_mutex_t cache_mutex;
uint32_t cache_size; uint32_t cache_size;
} WterlConnHandle; } WterlConnHandle;
@ -69,11 +70,11 @@ typedef struct {
struct wterl_event_handlers { struct wterl_event_handlers {
WT_EVENT_HANDLER handlers; WT_EVENT_HANDLER handlers;
ErlNifEnv *msg_env_error; ErlNifEnv *msg_env_error;
ErlNifMutex *error_mutex; pthread_mutex_t error_mutex;
ErlNifEnv *msg_env_message; ErlNifEnv *msg_env_message;
ErlNifMutex *message_mutex; pthread_mutex_t message_mutex;
ErlNifEnv *msg_env_progress; ErlNifEnv *msg_env_progress;
ErlNifMutex *progress_mutex; pthread_mutex_t progress_mutex;
ErlNifPid to_pid; ErlNifPid to_pid;
}; };
@ -204,7 +205,7 @@ __ctx_cache_evict(WterlConnHandle *conn_handle)
STAILQ_REMOVE(&conn_handle->cache, c, wterl_ctx, entries); STAILQ_REMOVE(&conn_handle->cache, c, wterl_ctx, entries);
if (c->session) if (c->session)
c->session->close(c->session, NULL); c->session->close(c->session, NULL);
enif_free(c); free(c);
num_evicted++; num_evicted++;
} }
} }
@ -226,7 +227,7 @@ __ctx_cache_find(WterlConnHandle *conn_handle, const uint64_t sig)
{ {
struct wterl_ctx *c; struct wterl_ctx *c;
enif_mutex_lock(conn_handle->cache_mutex); pthread_mutex_lock(&conn_handle->cache_mutex);
c = STAILQ_FIRST(&conn_handle->cache); c = STAILQ_FIRST(&conn_handle->cache);
while (c != NULL) { while (c != NULL) {
if (c->sig == sig) { // TODO: hash collisions *will* lead to SEGVs if (c->sig == sig) { // TODO: hash collisions *will* lead to SEGVs
@ -237,7 +238,7 @@ __ctx_cache_find(WterlConnHandle *conn_handle, const uint64_t sig)
} }
c = STAILQ_NEXT(c, entries); c = STAILQ_NEXT(c, entries);
} }
enif_mutex_unlock(conn_handle->cache_mutex); pthread_mutex_unlock(&conn_handle->cache_mutex);
DPRINTF("cache_find: [%u] %s (%p)", conn_handle->cache_size, c ? "hit" : "miss", c); DPRINTF("cache_find: [%u] %s (%p)", conn_handle->cache_size, c ? "hit" : "miss", c);
return c; return c;
} }
@ -251,7 +252,7 @@ __ctx_cache_find(WterlConnHandle *conn_handle, const uint64_t sig)
static void static void
__ctx_cache_add(WterlConnHandle *conn_handle, struct wterl_ctx *c) __ctx_cache_add(WterlConnHandle *conn_handle, struct wterl_ctx *c)
{ {
enif_mutex_lock(conn_handle->cache_mutex); pthread_mutex_lock(&conn_handle->cache_mutex);
__ctx_cache_evict(conn_handle); __ctx_cache_evict(conn_handle);
STAILQ_INSERT_TAIL(&conn_handle->cache, c, entries); STAILQ_INSERT_TAIL(&conn_handle->cache, c, entries);
conn_handle->cache_size += 1; conn_handle->cache_size += 1;
@ -262,7 +263,7 @@ __ctx_cache_add(WterlConnHandle *conn_handle, struct wterl_ctx *c)
sz++; sz++;
} }
#endif #endif
enif_mutex_unlock(conn_handle->cache_mutex); pthread_mutex_unlock(&conn_handle->cache_mutex);
DPRINTF("cache_add: [%u:%u] (%p)", sz, conn_handle->cache_size, c); DPRINTF("cache_add: [%u:%u] (%p)", sz, conn_handle->cache_size, c);
} }
@ -333,7 +334,7 @@ __retain_ctx(WterlConnHandle *conn_handle, uint32_t worker_id,
int rc = conn->open_session(conn, NULL, session_config, &session); int rc = conn->open_session(conn, NULL, session_config, &session);
if (rc != 0) return rc; if (rc != 0) return rc;
size_t s = sizeof(struct wterl_ctx) + (count * sizeof(struct cursor_info)) + sig_len; size_t s = sizeof(struct wterl_ctx) + (count * sizeof(struct cursor_info)) + sig_len;
c = enif_alloc(s); // TODO: enif_alloc_resource() c = malloc(s); // TODO: enif_alloc_resource()
if (c == NULL) { if (c == NULL) {
session->close(session, NULL); session->close(session, NULL);
return ENOMEM; return ENOMEM;
@ -355,7 +356,7 @@ __retain_ctx(WterlConnHandle *conn_handle, uint32_t worker_id,
c->ci[i].config = __copy_str_into(&p, config); c->ci[i].config = __copy_str_into(&p, config);
rc = session->open_cursor(session, uri, NULL, config, &c->ci[i].cursor); rc = session->open_cursor(session, uri, NULL, config, &c->ci[i].cursor);
if (rc != 0) { if (rc != 0) {
enif_free(c); free(c);
session->close(session, NULL); // this will free the cursors too session->close(session, NULL); // this will free the cursors too
va_end(ap); va_end(ap);
return rc; return rc;
@ -391,7 +392,7 @@ __release_ctx(WterlConnHandle *conn_handle, uint32_t worker_id, struct wterl_ctx
/** /**
* Close all sessions and all cursors open on any objects. * Close all sessions and all cursors open on any objects.
* *
* Note: always call within enif_mutex_lock/unlock(conn_handle->cache_mutex) * Note: always call within pthread_mutex_lock/unlock(conn_handle->cache_mutex)
*/ */
void void
__close_all_sessions(WterlConnHandle *conn_handle) __close_all_sessions(WterlConnHandle *conn_handle)
@ -405,7 +406,7 @@ __close_all_sessions(WterlConnHandle *conn_handle)
STAILQ_REMOVE(&conn_handle->cache, c, wterl_ctx, entries); STAILQ_REMOVE(&conn_handle->cache, c, wterl_ctx, entries);
conn_handle->cache_size -= 1; conn_handle->cache_size -= 1;
c->session->close(c->session, NULL); c->session->close(c->session, NULL);
enif_free(c); free(c);
c = n; c = n;
} }
} }
@ -413,7 +414,7 @@ __close_all_sessions(WterlConnHandle *conn_handle)
/** /**
* Close cursors open on 'uri' object. * Close cursors open on 'uri' object.
* *
* Note: always call within enif_mutex_lock/unlock(conn_handle->cache_mutex) * Note: always call within pthread_mutex_lock/unlock(conn_handle->cache_mutex)
*/ */
void void
__close_cursors_on(WterlConnHandle *conn_handle, const char *uri) __close_cursors_on(WterlConnHandle *conn_handle, const char *uri)
@ -431,7 +432,7 @@ __close_cursors_on(WterlConnHandle *conn_handle, const char *uri)
STAILQ_REMOVE(&conn_handle->cache, c, wterl_ctx, entries); STAILQ_REMOVE(&conn_handle->cache, c, wterl_ctx, entries);
conn_handle->cache_size -= 1; conn_handle->cache_size -= 1;
c->session->close(c->session, NULL); c->session->close(c->session, NULL);
enif_free(c); free(c);
break; break;
} }
} }
@ -461,7 +462,7 @@ __wterl_error_handler(WT_EVENT_HANDLER *handler, int error, const char *message)
ErlNifPid *to_pid; ErlNifPid *to_pid;
int rc = 0; int rc = 0;
enif_mutex_lock(eh->error_mutex); pthread_mutex_lock(&eh->error_mutex);
msg_env = eh->msg_env_error; msg_env = eh->msg_env_error;
to_pid = &eh->to_pid; to_pid = &eh->to_pid;
if (msg_env) { if (msg_env) {
@ -476,7 +477,7 @@ __wterl_error_handler(WT_EVENT_HANDLER *handler, int error, const char *message)
} else { } else {
rc = (fprintf(stderr, "[%d] %s\n", error, message) >= 0 ? 0 : EIO); rc = (fprintf(stderr, "[%d] %s\n", error, message) >= 0 ? 0 : EIO);
} }
enif_mutex_unlock(eh->error_mutex); pthread_mutex_unlock(&eh->error_mutex);
return rc; return rc;
} }
@ -499,7 +500,7 @@ __wterl_message_handler(WT_EVENT_HANDLER *handler, const char *message)
ErlNifPid *to_pid; ErlNifPid *to_pid;
int rc = 0; int rc = 0;
enif_mutex_lock(eh->message_mutex); pthread_mutex_lock(&eh->message_mutex);
msg_env = eh->msg_env_message; msg_env = eh->msg_env_message;
to_pid = &eh->to_pid; to_pid = &eh->to_pid;
if (msg_env) { if (msg_env) {
@ -512,7 +513,7 @@ __wterl_message_handler(WT_EVENT_HANDLER *handler, const char *message)
} else { } else {
rc = (printf("%s\n", message) >= 0 ? 0 : EIO); rc = (printf("%s\n", message) >= 0 ? 0 : EIO);
} }
enif_mutex_unlock(eh->message_mutex); pthread_mutex_unlock(&eh->message_mutex);
return rc; return rc;
} }
@ -536,7 +537,7 @@ __wterl_progress_handler(WT_EVENT_HANDLER *handler, const char *operation, uint6
ErlNifPid *to_pid; ErlNifPid *to_pid;
int rc = 0; int rc = 0;
enif_mutex_lock(eh->progress_mutex); pthread_mutex_lock(&eh->progress_mutex);
msg_env = eh->msg_env_progress; msg_env = eh->msg_env_progress;
to_pid = &eh->to_pid; to_pid = &eh->to_pid;
if (msg_env) { if (msg_env) {
@ -551,7 +552,7 @@ __wterl_progress_handler(WT_EVENT_HANDLER *handler, const char *operation, uint6
} else { } else {
rc = (printf("[%llu] %s\n", PRIuint64(counter), operation) >= 0 ? 0 : EIO); rc = (printf("[%llu] %s\n", PRIuint64(counter), operation) >= 0 ? 0 : EIO);
} }
enif_mutex_unlock(eh->progress_mutex); pthread_mutex_unlock(&eh->progress_mutex);
return rc; return rc;
} }
@ -637,7 +638,7 @@ ASYNC_NIF_DECL(
return; return;
} }
if (session_config.size > 1) { if (session_config.size > 1) {
char *sc = enif_alloc(session_config.size); char *sc = malloc(session_config.size);
if (!sc) { if (!sc) {
enif_release_resource(conn_handle); enif_release_resource(conn_handle);
ASYNC_NIF_REPLY(__strerror_term(env, ENOMEM)); ASYNC_NIF_REPLY(__strerror_term(env, ENOMEM));
@ -648,8 +649,8 @@ ASYNC_NIF_DECL(
} else { } else {
conn_handle->session_config = NULL; conn_handle->session_config = NULL;
} }
conn_handle->cache_mutex = enif_mutex_create("conn_handle"); pthread_mutex_init(&conn_handle->cache_mutex, 0);
enif_mutex_lock(conn_handle->cache_mutex); pthread_mutex_lock(&conn_handle->cache_mutex);
conn_handle->conn = conn; conn_handle->conn = conn;
ERL_NIF_TERM result = enif_make_resource(env, conn_handle); ERL_NIF_TERM result = enif_make_resource(env, conn_handle);
@ -658,7 +659,7 @@ ASYNC_NIF_DECL(
conn_handle->cache_size = 0; conn_handle->cache_size = 0;
enif_release_resource(conn_handle); enif_release_resource(conn_handle);
enif_mutex_unlock(conn_handle->cache_mutex); pthread_mutex_unlock(&conn_handle->cache_mutex);
ASYNC_NIF_REPLY(enif_make_tuple2(env, ATOM_OK, result)); ASYNC_NIF_REPLY(enif_make_tuple2(env, ATOM_OK, result));
} }
else else
@ -695,16 +696,16 @@ ASYNC_NIF_DECL(
{ // work { // work
/* Free up the shared sessions and cursors. */ /* Free up the shared sessions and cursors. */
enif_mutex_lock(args->conn_handle->cache_mutex); pthread_mutex_lock(&args->conn_handle->cache_mutex);
__close_all_sessions(args->conn_handle); __close_all_sessions(args->conn_handle);
if (args->conn_handle->session_config) { if (args->conn_handle->session_config) {
enif_free((char *)args->conn_handle->session_config); free((char *)args->conn_handle->session_config);
args->conn_handle->session_config = NULL; args->conn_handle->session_config = NULL;
} }
WT_CONNECTION* conn = args->conn_handle->conn; WT_CONNECTION* conn = args->conn_handle->conn;
int rc = conn->close(conn, NULL); int rc = conn->close(conn, NULL);
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
enif_mutex_destroy(args->conn_handle->cache_mutex); pthread_mutex_destroy(&args->conn_handle->cache_mutex);
memset(args->conn_handle, 0, sizeof(WterlConnHandle)); memset(args->conn_handle, 0, sizeof(WterlConnHandle));
ASYNC_NIF_REPLY(rc == 0 ? ATOM_OK : __strerror_term(env, rc)); ASYNC_NIF_REPLY(rc == 0 ? ATOM_OK : __strerror_term(env, rc));
@ -800,12 +801,12 @@ ASYNC_NIF_DECL(
{ // work { // work
/* This call requires that there be no open cursors referencing the object. */ /* This call requires that there be no open cursors referencing the object. */
enif_mutex_lock(args->conn_handle->cache_mutex); pthread_mutex_lock(&args->conn_handle->cache_mutex);
__close_cursors_on(args->conn_handle, args->uri); __close_cursors_on(args->conn_handle, args->uri);
ErlNifBinary config; ErlNifBinary config;
if (!enif_inspect_binary(env, args->config, &config)) { if (!enif_inspect_binary(env, args->config, &config)) {
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(enif_make_badarg(env)); ASYNC_NIF_REPLY(enif_make_badarg(env));
return; return;
} }
@ -817,7 +818,7 @@ ASYNC_NIF_DECL(
WT_SESSION *session = NULL; WT_SESSION *session = NULL;
int rc = conn->open_session(conn, NULL, args->conn_handle->session_config, &session); int rc = conn->open_session(conn, NULL, args->conn_handle->session_config, &session);
if (rc != 0) { if (rc != 0) {
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(__strerror_term(env, rc)); ASYNC_NIF_REPLY(__strerror_term(env, rc));
return; return;
} }
@ -827,7 +828,7 @@ ASYNC_NIF_DECL(
this will result in EBUSY(16) "Device or resource busy". */ this will result in EBUSY(16) "Device or resource busy". */
rc = session->drop(session, args->uri, (const char*)config.data); rc = session->drop(session, args->uri, (const char*)config.data);
(void)session->close(session, NULL); (void)session->close(session, NULL);
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(rc == 0 ? ATOM_OK : __strerror_term(env, rc)); ASYNC_NIF_REPLY(rc == 0 ? ATOM_OK : __strerror_term(env, rc));
}, },
{ // post { // post
@ -867,12 +868,12 @@ ASYNC_NIF_DECL(
{ // work { // work
/* This call requires that there be no open cursors referencing the object. */ /* This call requires that there be no open cursors referencing the object. */
enif_mutex_lock(args->conn_handle->cache_mutex); pthread_mutex_lock(&args->conn_handle->cache_mutex);
__close_cursors_on(args->conn_handle, args->oldname); __close_cursors_on(args->conn_handle, args->oldname);
ErlNifBinary config; ErlNifBinary config;
if (!enif_inspect_binary(env, args->config, &config)) { if (!enif_inspect_binary(env, args->config, &config)) {
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(enif_make_badarg(env)); ASYNC_NIF_REPLY(enif_make_badarg(env));
return; return;
} }
@ -884,7 +885,7 @@ ASYNC_NIF_DECL(
WT_SESSION *session = NULL; WT_SESSION *session = NULL;
int rc = conn->open_session(conn, NULL, args->conn_handle->session_config, &session); int rc = conn->open_session(conn, NULL, args->conn_handle->session_config, &session);
if (rc != 0) { if (rc != 0) {
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(__strerror_term(env, rc)); ASYNC_NIF_REPLY(__strerror_term(env, rc));
return; return;
} }
@ -895,7 +896,7 @@ ASYNC_NIF_DECL(
this will result in EBUSY(16) "Device or resource busy". */ this will result in EBUSY(16) "Device or resource busy". */
rc = session->rename(session, args->oldname, args->newname, (const char*)config.data); rc = session->rename(session, args->oldname, args->newname, (const char*)config.data);
(void)session->close(session, NULL); (void)session->close(session, NULL);
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(rc == 0 ? ATOM_OK : __strerror_term(env, rc)); ASYNC_NIF_REPLY(rc == 0 ? ATOM_OK : __strerror_term(env, rc));
}, },
{ // post { // post
@ -935,12 +936,12 @@ ASYNC_NIF_DECL(
{ // work { // work
/* This call requires that there be no open cursors referencing the object. */ /* This call requires that there be no open cursors referencing the object. */
enif_mutex_lock(args->conn_handle->cache_mutex); pthread_mutex_lock(&args->conn_handle->cache_mutex);
__close_cursors_on(args->conn_handle, args->uri); __close_cursors_on(args->conn_handle, args->uri);
ErlNifBinary config; ErlNifBinary config;
if (!enif_inspect_binary(env, args->config, &config)) { if (!enif_inspect_binary(env, args->config, &config)) {
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(enif_make_badarg(env)); ASYNC_NIF_REPLY(enif_make_badarg(env));
return; return;
} }
@ -952,14 +953,14 @@ ASYNC_NIF_DECL(
WT_SESSION *session = NULL; WT_SESSION *session = NULL;
int rc = conn->open_session(conn, NULL, args->conn_handle->session_config, &session); int rc = conn->open_session(conn, NULL, args->conn_handle->session_config, &session);
if (rc != 0) { if (rc != 0) {
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(__strerror_term(env, rc)); ASYNC_NIF_REPLY(__strerror_term(env, rc));
return; return;
} }
rc = session->salvage(session, args->uri, (const char*)config.data); rc = session->salvage(session, args->uri, (const char*)config.data);
(void)session->close(session, NULL); (void)session->close(session, NULL);
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(rc == 0 ? ATOM_OK : __strerror_term(env, rc)); ASYNC_NIF_REPLY(rc == 0 ? ATOM_OK : __strerror_term(env, rc));
}, },
{ // post { // post
@ -1067,12 +1068,12 @@ ASYNC_NIF_DECL(
{ // work { // work
/* This call requires that there be no open cursors referencing the object. */ /* This call requires that there be no open cursors referencing the object. */
enif_mutex_lock(args->conn_handle->cache_mutex); pthread_mutex_lock(&args->conn_handle->cache_mutex);
__close_cursors_on(args->conn_handle, args->uri); __close_cursors_on(args->conn_handle, args->uri);
ErlNifBinary config; ErlNifBinary config;
if (!enif_inspect_binary(env, args->config, &config)) { if (!enif_inspect_binary(env, args->config, &config)) {
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(enif_make_badarg(env)); ASYNC_NIF_REPLY(enif_make_badarg(env));
return; return;
} }
@ -1085,7 +1086,7 @@ ASYNC_NIF_DECL(
WT_SESSION *session = NULL; WT_SESSION *session = NULL;
int rc = conn->open_session(conn, NULL, args->conn_handle->session_config, &session); int rc = conn->open_session(conn, NULL, args->conn_handle->session_config, &session);
if (rc != 0) { if (rc != 0) {
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(__strerror_term(env, rc)); ASYNC_NIF_REPLY(__strerror_term(env, rc));
return; return;
} }
@ -1100,7 +1101,7 @@ ASYNC_NIF_DECL(
mess. */ mess. */
if (!args->from_first) { if (!args->from_first) {
if (!enif_inspect_binary(env, args->start, &start_key)) { if (!enif_inspect_binary(env, args->start, &start_key)) {
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(enif_make_badarg(env)); ASYNC_NIF_REPLY(enif_make_badarg(env));
return; return;
} }
@ -1108,7 +1109,7 @@ ASYNC_NIF_DECL(
rc = session->open_cursor(session, args->uri, NULL, "raw", &start); rc = session->open_cursor(session, args->uri, NULL, "raw", &start);
if (rc != 0) { if (rc != 0) {
session->close(session, NULL); session->close(session, NULL);
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(__strerror_term(env, rc)); ASYNC_NIF_REPLY(__strerror_term(env, rc));
return; return;
} }
@ -1118,7 +1119,7 @@ ASYNC_NIF_DECL(
if (rc != 0) { if (rc != 0) {
start->close(start); start->close(start);
session->close(session, NULL); session->close(session, NULL);
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(__strerror_term(env, rc)); ASYNC_NIF_REPLY(__strerror_term(env, rc));
return; return;
} }
@ -1133,7 +1134,7 @@ ASYNC_NIF_DECL(
if (!enif_inspect_binary(env, args->stop, &stop_key)) { if (!enif_inspect_binary(env, args->stop, &stop_key)) {
start->close(start); start->close(start);
session->close(session, NULL); session->close(session, NULL);
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(enif_make_badarg(env)); ASYNC_NIF_REPLY(enif_make_badarg(env));
return; return;
} }
@ -1142,7 +1143,7 @@ ASYNC_NIF_DECL(
if (rc != 0) { if (rc != 0) {
start->close(start); start->close(start);
session->close(session, NULL); session->close(session, NULL);
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(__strerror_term(env, rc)); ASYNC_NIF_REPLY(__strerror_term(env, rc));
return; return;
} }
@ -1153,7 +1154,7 @@ ASYNC_NIF_DECL(
start->close(start); start->close(start);
stop->close(stop); stop->close(stop);
session->close(session, NULL); session->close(session, NULL);
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(__strerror_term(env, rc)); ASYNC_NIF_REPLY(__strerror_term(env, rc));
return; return;
} }
@ -1171,7 +1172,7 @@ ASYNC_NIF_DECL(
start->close(start); start->close(start);
stop->close(stop); stop->close(stop);
session->close(session, NULL); session->close(session, NULL);
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(rc == 0 ? ATOM_OK : __strerror_term(env, rc)); ASYNC_NIF_REPLY(rc == 0 ? ATOM_OK : __strerror_term(env, rc));
}, },
{ // post { // post
@ -1208,12 +1209,12 @@ ASYNC_NIF_DECL(
{ // work { // work
/* This call requires that there be no open cursors referencing the object. */ /* This call requires that there be no open cursors referencing the object. */
enif_mutex_lock(args->conn_handle->cache_mutex); pthread_mutex_lock(&args->conn_handle->cache_mutex);
__close_cursors_on(args->conn_handle, args->uri); __close_cursors_on(args->conn_handle, args->uri);
ErlNifBinary config; ErlNifBinary config;
if (!enif_inspect_binary(env, args->config, &config)) { if (!enif_inspect_binary(env, args->config, &config)) {
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(enif_make_badarg(env)); ASYNC_NIF_REPLY(enif_make_badarg(env));
return; return;
} }
@ -1225,14 +1226,14 @@ ASYNC_NIF_DECL(
WT_SESSION *session = NULL; WT_SESSION *session = NULL;
int rc = conn->open_session(conn, NULL, args->conn_handle->session_config, &session); int rc = conn->open_session(conn, NULL, args->conn_handle->session_config, &session);
if (rc != 0) { if (rc != 0) {
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(__strerror_term(env, rc)); ASYNC_NIF_REPLY(__strerror_term(env, rc));
return; return;
} }
rc = session->upgrade(session, args->uri, (const char*)config.data); rc = session->upgrade(session, args->uri, (const char*)config.data);
(void)session->close(session, NULL); (void)session->close(session, NULL);
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(rc == 0 ? ATOM_OK : __strerror_term(env, rc)); ASYNC_NIF_REPLY(rc == 0 ? ATOM_OK : __strerror_term(env, rc));
}, },
{ // post { // post
@ -1270,12 +1271,12 @@ ASYNC_NIF_DECL(
{ // work { // work
/* This call requires that there be no open cursors referencing the object. */ /* This call requires that there be no open cursors referencing the object. */
enif_mutex_lock(args->conn_handle->cache_mutex); pthread_mutex_lock(&args->conn_handle->cache_mutex);
__close_all_sessions(args->conn_handle); __close_all_sessions(args->conn_handle);
ErlNifBinary config; ErlNifBinary config;
if (!enif_inspect_binary(env, args->config, &config)) { if (!enif_inspect_binary(env, args->config, &config)) {
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(enif_make_badarg(env)); ASYNC_NIF_REPLY(enif_make_badarg(env));
return; return;
} }
@ -1287,14 +1288,14 @@ ASYNC_NIF_DECL(
WT_SESSION *session = NULL; WT_SESSION *session = NULL;
int rc = conn->open_session(conn, NULL, args->conn_handle->session_config, &session); int rc = conn->open_session(conn, NULL, args->conn_handle->session_config, &session);
if (rc != 0) { if (rc != 0) {
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(__strerror_term(env, rc)); ASYNC_NIF_REPLY(__strerror_term(env, rc));
return; return;
} }
rc = session->verify(session, args->uri, (const char*)config.data); rc = session->verify(session, args->uri, (const char*)config.data);
(void)session->close(session, NULL); (void)session->close(session, NULL);
enif_mutex_unlock(args->conn_handle->cache_mutex); pthread_mutex_unlock(&args->conn_handle->cache_mutex);
ASYNC_NIF_REPLY(rc == 0 ? ATOM_OK : __strerror_term(env, rc)); ASYNC_NIF_REPLY(rc == 0 ? ATOM_OK : __strerror_term(env, rc));
}, },
{ // post { // post
@ -2220,13 +2221,13 @@ static void __wterl_conn_dtor(ErlNifEnv* env, void* obj)
UNUSED(env); UNUSED(env);
WterlConnHandle *conn_handle = (WterlConnHandle *)obj; WterlConnHandle *conn_handle = (WterlConnHandle *)obj;
if (conn_handle->cache_mutex) { if (conn_handle->conn) {
DPRINTF("conn_handle dtor free'ing (%p)", obj); DPRINTF("conn_handle dtor free'ing (%p)", obj);
enif_mutex_lock(conn_handle->cache_mutex); pthread_mutex_lock(&conn_handle->cache_mutex);
__close_all_sessions(conn_handle); __close_all_sessions(conn_handle);
conn_handle->conn->close(conn_handle->conn, NULL); conn_handle->conn->close(conn_handle->conn, NULL);
enif_mutex_unlock(conn_handle->cache_mutex); pthread_mutex_unlock(&conn_handle->cache_mutex);
enif_mutex_destroy(conn_handle->cache_mutex); pthread_mutex_destroy(&conn_handle->cache_mutex);
} }
} }
@ -2265,15 +2266,15 @@ on_load(ErlNifEnv *env, void **priv_data, ERL_NIF_TERM load_info)
ATOM_WIREDTIGER_VSN = enif_make_atom(env, "wiredtiger_vsn"); ATOM_WIREDTIGER_VSN = enif_make_atom(env, "wiredtiger_vsn");
ATOM_MSG_PID = enif_make_atom(env, "message_pid"); ATOM_MSG_PID = enif_make_atom(env, "message_pid");
struct wterl_priv_data *priv = enif_alloc(sizeof(struct wterl_priv_data)); struct wterl_priv_data *priv = malloc(sizeof(struct wterl_priv_data));
if (!priv) if (!priv)
return ENOMEM; return ENOMEM;
memset(priv, 0, sizeof(struct wterl_priv_data)); memset(priv, 0, sizeof(struct wterl_priv_data));
struct wterl_event_handlers *eh = &priv->eh; struct wterl_event_handlers *eh = &priv->eh;
eh->error_mutex = enif_mutex_create("error_mutex"); pthread_mutex_init(&eh->error_mutex, 0);
eh->message_mutex = enif_mutex_create("message_mutex"); pthread_mutex_init(&eh->message_mutex, 0);
eh->progress_mutex = enif_mutex_create("progress_mutex"); pthread_mutex_init(&eh->progress_mutex, 0);
/* Process the load_info array of tuples, we expect: /* Process the load_info array of tuples, we expect:
[{wterl_vsn, "a version string"}, [{wterl_vsn, "a version string"},
@ -2296,7 +2297,7 @@ on_load(ErlNifEnv *env, void **priv_data, ERL_NIF_TERM load_info)
ASYNC_NIF_LOAD(wterl, priv->async_nif_priv); ASYNC_NIF_LOAD(wterl, priv->async_nif_priv);
if (!priv->async_nif_priv) { if (!priv->async_nif_priv) {
memset(priv, 0, sizeof(struct wterl_priv_data)); memset(priv, 0, sizeof(struct wterl_priv_data));
enif_free(priv); free(priv);
return ENOMEM; return ENOMEM;
} }
*priv_data = priv; *priv_data = priv;
@ -2331,9 +2332,9 @@ on_unload(ErlNifEnv *env, void *priv_data)
is no chance that the event handler functions will be called so we can is no chance that the event handler functions will be called so we can
be sure that there won't be a race on eh.msg_env in the callback functions. */ be sure that there won't be a race on eh.msg_env in the callback functions. */
struct wterl_event_handlers *eh = &priv->eh; struct wterl_event_handlers *eh = &priv->eh;
enif_mutex_destroy(eh->error_mutex); pthread_mutex_destroy(&eh->error_mutex);
enif_mutex_destroy(eh->message_mutex); pthread_mutex_destroy(&eh->message_mutex);
enif_mutex_destroy(eh->progress_mutex); pthread_mutex_destroy(&eh->progress_mutex);
if (eh->msg_env_message) if (eh->msg_env_message)
enif_free_env(eh->msg_env_message); enif_free_env(eh->msg_env_message);
if (eh->msg_env_error) if (eh->msg_env_error)
@ -2342,7 +2343,7 @@ on_unload(ErlNifEnv *env, void *priv_data)
enif_free_env(eh->msg_env_progress); enif_free_env(eh->msg_env_progress);
memset(priv, 0, sizeof(struct wterl_priv_data)); memset(priv, 0, sizeof(struct wterl_priv_data));
enif_free(priv); free(priv);
priv_data = NULL; priv_data = NULL;
} }