aboutsummaryrefslogtreecommitdiffstats
path: root/erts
diff options
context:
space:
mode:
Diffstat (limited to 'erts')
-rw-r--r--erts/doc/src/erl_nif.xml117
-rw-r--r--erts/emulator/beam/erl_alloc.c2
-rw-r--r--erts/emulator/beam/erl_alloc.types1
-rw-r--r--erts/emulator/beam/erl_driver.h9
-rw-r--r--erts/emulator/beam/erl_drv_nif.h15
-rw-r--r--erts/emulator/beam/erl_nif.c91
-rw-r--r--erts/emulator/beam/erl_nif.h19
-rw-r--r--erts/emulator/beam/erl_nif_api_funcs.h4
-rw-r--r--erts/emulator/beam/global.h29
-rw-r--r--erts/emulator/sys/common/erl_check_io.c890
-rw-r--r--erts/emulator/sys/common/erl_check_io.h18
-rw-r--r--erts/emulator/sys/common/erl_poll.c1
-rw-r--r--erts/emulator/sys/unix/sys.c10
-rw-r--r--erts/emulator/test/nif_SUITE.erl140
-rw-r--r--erts/emulator/test/nif_SUITE_data/nif_SUITE.c254
15 files changed, 1417 insertions, 183 deletions
diff --git a/erts/doc/src/erl_nif.xml b/erts/doc/src/erl_nif.xml
index 51b095e6ef..8cdfc80879 100644
--- a/erts/doc/src/erl_nif.xml
+++ b/erts/doc/src/erl_nif.xml
@@ -696,12 +696,32 @@ typedef struct {
Each resource type has a unique name and a destructor function that
is called when objects of its type are released.</p>
</item>
+ <tag><marker id="ErlNifResourceTypeInit"/><c>ErlNifResourceTypeInit</c></tag>
+ <item>
+ <code type="none">
+typedef struct {
+ ErlNifResourceDtor* dtor;
+ ErlNifResourceStop* stop;
+} ErlNifResourceTypeInit;</code>
+ <p>Initialization structure read by <seealso marker="#enif_open_resource_type_x">
+ enif_open_resource_type_x</seealso>.</p>
+ </item>
<tag><marker id="ErlNifResourceDtor"/><c>ErlNifResourceDtor</c></tag>
<item>
<code type="none">
typedef void ErlNifResourceDtor(ErlNifEnv* env, void* obj);</code>
<p>The function prototype of a resource destructor function.</p>
</item>
+ <tag><marker id="ErlNifResourceStop"/><c>ErlNifResourceStop</c></tag>
+ <item>
+ <code type="none">
+typedef void ErlNifResourceStop(ErlNifEnv* env, void* obj, ErlNifEvent event, int is_direct_call);</code>
+ <p>The function prototype of a resource stop function,
+ called on the behalf of <seealso marker="#enif_select">
+ enif_select</seealso>. <c>obj</c> is the resource, <c>event</c> is OS event,
+ <c>is_direct_call</c> is true if the call is made directly from <c>enif_select</c>
+ or false if it is a scheduled call (potentially from another thread).</p>
+ </item>
<tag><marker id="ErlNifCharEncoding"/><c>ErlNifCharEncoding</c></tag>
<item>
<code type="none">
@@ -2236,6 +2256,24 @@ enif_map_iterator_destroy(env, &amp;iter);</code>
</func>
<func>
+ <name><ret>ErlNifResourceType *</ret>
+ <nametext>enif_open_resource_type_x(ErlNifEnv* env, const char* name,
+ ErlNifResourceTypeInit* init,
+ ErlNifResourceFlags flags, ErlNifResourceFlags* tried)</nametext>
+ </name>
+ <fsummary>Create or takeover a resource type.</fsummary>
+ <desc>
+ <p>Same as <seealso marker="#enif_open_resource_type"><c>enif_open_resource_type</c></seealso>
+ except is also accept a <c>stop</c> callback for resource types that are
+ used together with <seealso marker="#enif_select"><c>enif_select</c></seealso>.</p>
+ <p>Argument <c>init</c> is a pointer to an
+ <seealso marker="#ErlNifResourceTypeInit"><c>ErlNifResourceTypeInit</c></seealso>
+ structure that contains the function pointers for the destructor and the stop callback
+ of the resource type.</p>
+ </desc>
+ </func>
+
+ <func>
<name><ret>int</ret><nametext>enif_port_command(ErlNifEnv* env, const
ErlNifPort* to_port, ErlNifEnv *msg_env, ERL_NIF_TERM msg)</nametext>
</name>
@@ -2342,7 +2380,7 @@ enif_map_iterator_destroy(env, &amp;iter);</code>
<nametext>enif_release_resource(void* obj)</nametext></name>
<fsummary>Release a resource object.</fsummary>
<desc>
- <p>Removes a reference to resource object <c>obj</c>obtained from
+ <p>Removes a reference to resource object <c>obj</c> obtained from
<seealso marker="#enif_alloc_resource">
<c>enif_alloc_resource</c></seealso>.
The resource object is destructed when the last reference is removed.
@@ -2482,6 +2520,83 @@ enif_map_iterator_destroy(env, &amp;iter);</code>
</func>
<func>
+ <name><ret>enum ErlNifSelectReturn</ret>
+ <nametext>enif_select(ErlNifEnv* env, ErlNifEvent event,
+ enum ErlNifSelectFlags mode, void* obj, Eterm ref)</nametext>
+ </name>
+ <fsummary>Manage subscription on IO event.</fsummary>
+ <desc>
+ <p>This function can be used to receive asynchronous notifications
+ when OS-specific event objects become ready for either read or write operations.</p>
+ <p>Argument <c>event</c> identifies the event object. On Unix
+ systems, the functions <c>select</c>/<c>poll</c> are used. The event
+ object must be a socket, pipe or other file descriptor object that
+ <c>select</c>/<c>poll</c> can use.</p>
+ <p>Argument <c>mode</c> describes the type of events to wait for. It can be
+ <c>ERL_NIF_SELECT_READ</c>, <c>ERL_NIF_SELECT_WRITE</c> or a bitwise
+ OR combination to wait for both. It can also be <c>ERL_NIF_SELECT_STOP</c>
+ which is described further below. When a read or write event is triggerred,
+ a notification message like this is sent to the Erlang process that called
+ <c>enif_select</c>:</p>
+ <code type="none"><c>{select, Obj, Ref, ready_input | ready_output}</c></code>
+ <p><c>ready_input</c> or <c>ready_output</c> indicates if the event object
+ is ready for reading or writing.</p>
+ <p>Argument <c>obj</c> is a resource object obtained from
+ <seealso marker="#enif_alloc_resource"><c>enif_alloc_resource</c></seealso>.
+ The purpose of the resource objects is as a container of the event object
+ to manage its state and lifetime. A handle to the resource is received
+ in the notification message as <c>Obj</c>.</p>
+ <p>Argument <c>ref</c> must be either a reference obtained from
+ <seealso marker="erlang#make_ref-0"><c>erlang:make_ref/0</c></seealso>
+ or the atom <c>undefined</c>. It will be passed as <c>Ref</c> in the notifications.
+ If a selective <c>receive</c> statement is used to wait for the notification
+ then a reference created just before the <c>receive</c> will exploit a runtime
+ optimization that bypasses all earlier received messages in the queue.</p>
+ <p>The notifications are one-shot only. To receive further notifications of the same
+ type (read or write), repeated calls to <c>enif_select</c> must be made.</p>
+ <p>Use <c>ERL_NIF_SELECT_STOP</c> as <c>mode</c> in order to safely
+ close an event object that has been passed to <c>enif_select</c>. The
+ <seealso marker="#ErlNifResourceStop"><c>stop</c></seealso> callback
+ of the resource <c>obj</c> will be called when it is safe to close
+ the event object. This safe way of closing event objects must be used
+ even if all notifications have been received and no further calls to
+ <c>enif_select</c> have been made.</p>
+ <p>Returns an integer where different bits indicate the outcome of the call:</p>
+ <taglist>
+ <tag><c>ERL_NIF_SELECT_ERROR</c></tag>
+ <item>The master error bit. It will always be set if the call failed for
+ any reason.</item>
+ <tag><c>ERL_NIF_SELECT_STOP_CALLED</c></tag>
+ <item>The stop callback was called directly by <c>enif_select</c>.</item>
+ <tag><c>ERL_NIF_SELECT_STOP_SCHEDULED</c></tag>
+ <item>The stop callback was scheduled to run on some other thread
+ or later by this thread.</item>
+ <tag><c>ERL_NIF_SELECT_INVALID_EVENT</c></tag>
+ <item>Argument <c>event</c> is not a valid OS event object.</item>
+ <tag><c>ERL_NIF_SELECT_FAILED</c></tag>
+ <item>The system call failed to add the event object to the poll set.</item>
+ </taglist>
+ <p>The return value from a successful call with <c>mode</c> as <c>ERL_NIF_SELECT_STOP</c>,
+ will contain either bit <c>ERL_NIF_SELECT_STOP_CALLED</c> or
+ <c>ERL_NIF_SELECT_STOP_SCHEDULED</c>.</p>
+ <note>
+ <p>Always use bitwise AND to test the return value. New significant bits
+ may be added in future releases to give more detailed information for both
+ failed and successful calls. Do NOT use equallity tests like <c>==</c>, as
+ that may cause your application to stop working.</p>
+ <p>Example:</p>
+ <code type="none">
+retval = enif_select(env, fd, ERL_NIF_SELECT_READ, resource, ref);
+if (retval &amp; ERL_NIF_SELECT_ERROR) {
+ /* handle error */
+}
+/* Success! */
+</code>
+ </note>
+ </desc>
+ </func>
+
+ <func>
<name><ret>ErlNifPid *</ret>
<nametext>enif_self(ErlNifEnv* caller_env, ErlNifPid* pid)</nametext>
</name>
diff --git a/erts/emulator/beam/erl_alloc.c b/erts/emulator/beam/erl_alloc.c
index 4d990a9c56..43c22e2e7c 100644
--- a/erts/emulator/beam/erl_alloc.c
+++ b/erts/emulator/beam/erl_alloc.c
@@ -664,6 +664,8 @@ erts_alloc_init(int *argc, char **argv, ErtsAllocInitOpts *eaiop)
= sizeof(ErtsDrvEventDataState);
fix_type_sizes[ERTS_ALC_FIX_TYPE_IX(ERTS_ALC_T_DRV_SEL_D_STATE)]
= sizeof(ErtsDrvSelectDataState);
+ fix_type_sizes[ERTS_ALC_FIX_TYPE_IX(ERTS_ALC_T_NIF_SEL_D_STATE)]
+ = sizeof(ErtsNifSelectDataState);
fix_type_sizes[ERTS_ALC_FIX_TYPE_IX(ERTS_ALC_T_MSG_REF)]
= sizeof(ErtsMessageRef);
#ifdef ERTS_SMP
diff --git a/erts/emulator/beam/erl_alloc.types b/erts/emulator/beam/erl_alloc.types
index 70eca5b49c..1f7e4400e1 100644
--- a/erts/emulator/beam/erl_alloc.types
+++ b/erts/emulator/beam/erl_alloc.types
@@ -397,6 +397,7 @@ type DRV_TAB LONG_LIVED SYSTEM drv_tab
type DRV_EV_STATE LONG_LIVED SYSTEM driver_event_state
type DRV_EV_D_STATE FIXED_SIZE SYSTEM driver_event_data_state
type DRV_SEL_D_STATE FIXED_SIZE SYSTEM driver_select_data_state
+type NIF_SEL_D_STATE FIXED_SIZE SYSTEM enif_select_data_state
type FD_LIST SHORT_LIVED SYSTEM fd_list
type ACTIVE_FD_ARR SHORT_LIVED SYSTEM active_fd_array
type POLLSET LONG_LIVED SYSTEM pollset
diff --git a/erts/emulator/beam/erl_driver.h b/erts/emulator/beam/erl_driver.h
index 97a69140c3..5bea92e198 100644
--- a/erts/emulator/beam/erl_driver.h
+++ b/erts/emulator/beam/erl_driver.h
@@ -76,11 +76,10 @@ typedef struct {
# endif
#endif
-/* Values for mode arg to driver_select() */
-#define ERL_DRV_READ (1 << 0)
-#define ERL_DRV_WRITE (1 << 1)
-#define ERL_DRV_USE (1 << 2)
-#define ERL_DRV_USE_NO_CALLBACK (ERL_DRV_USE | (1 << 3))
+#define ERL_DRV_READ ((int)ERL_NIF_SELECT_READ)
+#define ERL_DRV_WRITE ((int)ERL_NIF_SELECT_WRITE)
+#define ERL_DRV_USE ((int)ERL_NIF_SELECT_STOP)
+#define ERL_DRV_USE_NO_CALLBACK (ERL_DRV_USE | (ERL_DRV_USE << 1))
/* Old deprecated */
#define DO_READ ERL_DRV_READ
diff --git a/erts/emulator/beam/erl_drv_nif.h b/erts/emulator/beam/erl_drv_nif.h
index 2489099b5c..46bb06d642 100644
--- a/erts/emulator/beam/erl_drv_nif.h
+++ b/erts/emulator/beam/erl_drv_nif.h
@@ -49,6 +49,21 @@ typedef enum {
ERL_DIRTY_JOB_IO_BOUND = 2
} ErlDirtyJobFlags;
+/* Values for enif_select AND mode arg for driver_select() */
+enum ErlNifSelectFlags {
+ ERL_NIF_SELECT_READ = (1 << 0),
+ ERL_NIF_SELECT_WRITE = (1 << 1),
+ ERL_NIF_SELECT_STOP = (1 << 2)
+};
+
+enum ErlNifSelectReturn {
+ ERL_NIF_SELECT_ERROR = (1 << 0),
+ ERL_NIF_SELECT_STOP_CALLED = (1 << 1),
+ ERL_NIF_SELECT_STOP_SCHEDULED = (1 << 2),
+ ERL_NIF_SELECT_INVALID_EVENT = (1 << 3),
+ ERL_NIF_SELECT_FAILED = (1 << 4)
+};
+
#ifdef SIZEOF_CHAR
# define SIZEOF_CHAR_SAVED__ SIZEOF_CHAR
# undef SIZEOF_CHAR
diff --git a/erts/emulator/beam/erl_nif.c b/erts/emulator/beam/erl_nif.c
index b860759fa2..85fa53a886 100644
--- a/erts/emulator/beam/erl_nif.c
+++ b/erts/emulator/beam/erl_nif.c
@@ -1910,37 +1910,10 @@ int enif_snprintf(char *buffer, size_t size, const char* format, ...)
** Memory managed (GC'ed) "resource" objects **
***********************************************************/
-
-struct enif_resource_type_t
-{
- struct enif_resource_type_t* next; /* list of all resource types */
- struct enif_resource_type_t* prev;
- struct erl_module_nif* owner; /* that created this type and thus implements the destructor*/
- ErlNifResourceDtor* dtor; /* user destructor function */
- erts_refc_t refc; /* num of resources of this type (HOTSPOT warning)
- +1 for active erl_module_nif */
- Eterm module;
- Eterm name;
-};
-
/* dummy node in circular list */
struct enif_resource_type_t resource_type_list;
-typedef struct enif_resource_t
-{
- struct enif_resource_type_t* type;
-#ifdef DEBUG
- erts_refc_t nif_refc;
-# ifdef ARCH_32
- byte align__[4];
-# endif
-#endif
-
- char data[1];
-}ErlNifResource;
-
#define SIZEOF_ErlNifResource(SIZE) (offsetof(ErlNifResource,data) + (SIZE))
-#define DATA_TO_RESOURCE(PTR) ((ErlNifResource*)((char*)(PTR) - offsetof(ErlNifResource,data)))
static ErlNifResourceType* find_resource_type(Eterm module, Eterm name)
{
@@ -2003,24 +1976,23 @@ struct opened_resource_type
ErlNifResourceFlags op;
ErlNifResourceType* type;
- ErlNifResourceDtor* new_dtor;
+ ErlNifResourceTypeInit new_callbacks;
};
static struct opened_resource_type* opened_rt_list = NULL;
-ErlNifResourceType*
-enif_open_resource_type(ErlNifEnv* env,
- const char* module_str,
- const char* name_str,
- ErlNifResourceDtor* dtor,
- ErlNifResourceFlags flags,
- ErlNifResourceFlags* tried)
+static
+ErlNifResourceType* open_resource_type(ErlNifEnv* env,
+ const char* name_str,
+ const ErlNifResourceTypeInit* init,
+ ErlNifResourceFlags flags,
+ ErlNifResourceFlags* tried,
+ size_t sizeof_init)
{
ErlNifResourceType* type = NULL;
ErlNifResourceFlags op = flags;
Eterm module_am, name_am;
ASSERT(erts_smp_thr_progress_is_blocking());
- ASSERT(module_str == NULL); /* for now... */
module_am = make_atom(env->mod_nif->mod->module);
name_am = enif_make_atom(env, name_str);
@@ -2054,7 +2026,9 @@ enif_open_resource_type(ErlNifEnv* env,
sizeof(struct opened_resource_type));
ort->op = op;
ort->type = type;
- ort->new_dtor = dtor;
+ sys_memzero(&ort->new_callbacks, sizeof(ErlNifResourceTypeInit));
+ ASSERT(sizeof_init > 0 && sizeof_init <= sizeof(ErlNifResourceTypeInit));
+ sys_memcpy(&ort->new_callbacks, init, sizeof_init);
ort->next = opened_rt_list;
opened_rt_list = ort;
}
@@ -2064,6 +2038,31 @@ enif_open_resource_type(ErlNifEnv* env,
return type;
}
+ErlNifResourceType*
+enif_open_resource_type(ErlNifEnv* env,
+ const char* module_str,
+ const char* name_str,
+ ErlNifResourceDtor* dtor,
+ ErlNifResourceFlags flags,
+ ErlNifResourceFlags* tried)
+{
+ ErlNifResourceTypeInit init = {dtor, NULL};
+ ASSERT(module_str == NULL); /* for now... */
+ return open_resource_type(env, name_str, &init, flags, tried,
+ sizeof(init));
+}
+
+ErlNifResourceType*
+enif_open_resource_type_x(ErlNifEnv* env,
+ const char* name_str,
+ const ErlNifResourceTypeInit* init,
+ ErlNifResourceFlags flags,
+ ErlNifResourceFlags* tried)
+{
+ return open_resource_type(env, name_str, init, flags, tried,
+ env->mod_nif->entry.sizeof_ErlNifResourceTypeInit);
+}
+
static void commit_opened_resource_types(struct erl_module_nif* lib)
{
while (opened_rt_list) {
@@ -2082,7 +2081,8 @@ static void commit_opened_resource_types(struct erl_module_nif* lib)
}
type->owner = lib;
- type->dtor = ort->new_dtor;
+ type->dtor = ort->new_callbacks.dtor;
+ type->stop = ort->new_callbacks.stop;
if (type->dtor != NULL) {
erts_refc_inc(&lib->rt_dtor_cnt, 1);
@@ -2130,6 +2130,16 @@ static void nif_resource_dtor(Binary* bin)
}
}
+void erts_resource_stop(ErlNifResource* resource, ErlNifEvent e,
+ int is_direct_call)
+{
+ struct enif_msg_environment_t msg_env;
+ ASSERT(resource->type->stop);
+ pre_nif_noproc(&msg_env, resource->type->owner, NULL);
+ resource->type->stop(&msg_env.env, resource->data, e, is_direct_call);
+ post_nif_noproc(&msg_env);
+}
+
void* enif_alloc_resource(ErlNifResourceType* type, size_t size)
{
Binary* bin = erts_create_magic_binary_x(SIZEOF_ErlNifResource(size),
@@ -2969,6 +2979,11 @@ static struct erl_module_nif* create_lib(const ErlNifEntry* src)
dst->funcs = lib->_funcs_copy_;
dst->options = 0;
}
+ if (AT_LEAST_VERSION(src, 2, 12)) {
+ dst->sizeof_ErlNifResourceTypeInit = src->sizeof_ErlNifResourceTypeInit;
+ } else {
+ dst->sizeof_ErlNifResourceTypeInit = 0;
+ }
return lib;
};
diff --git a/erts/emulator/beam/erl_nif.h b/erts/emulator/beam/erl_nif.h
index 413b4f7343..78e0fa1864 100644
--- a/erts/emulator/beam/erl_nif.h
+++ b/erts/emulator/beam/erl_nif.h
@@ -52,7 +52,7 @@
** 2.11: 19.0 enif_snprintf
*/
#define ERL_NIF_MAJOR_VERSION 2
-#define ERL_NIF_MINOR_VERSION 11
+#define ERL_NIF_MINOR_VERSION 12
/*
* The emulator will refuse to load a nif-lib with a major version
@@ -122,6 +122,9 @@ typedef struct enif_entry_t
/* Added in 2.7 */
unsigned options; /* Unused. Can be set to 0 or 1 (dirty sched config) */
+
+ /* Added in 2.12 */
+ size_t sizeof_ErlNifResourceTypeInit;
}ErlNifEntry;
@@ -135,8 +138,19 @@ typedef struct
void* ref_bin;
}ErlNifBinary;
+//#ifndef ERL_SYS_DRV
+typedef int ErlNifEvent; /* An event to be selected on. */
+//#endif
+
typedef struct enif_resource_type_t ErlNifResourceType;
typedef void ErlNifResourceDtor(ErlNifEnv*, void*);
+typedef void ErlNifResourceStop(ErlNifEnv*, void*, ErlNifEvent, int is_direct_call);
+
+typedef struct {
+ ErlNifResourceDtor* dtor;
+ ErlNifResourceStop* stop; /* at ERL_NIF_SELECT_STOP event */
+} ErlNifResourceTypeInit;
+
typedef enum
{
ERL_NIF_RT_CREATE = 1,
@@ -292,7 +306,8 @@ ERL_NIF_INIT_DECL(NAME) \
FUNCS, \
LOAD, RELOAD, UPGRADE, UNLOAD, \
ERL_NIF_VM_VARIANT, \
- 1 \
+ 1, \
+ sizeof(ErlNifResourceTypeInit) \
}; \
ERL_NIF_INIT_BODY; \
return &entry; \
diff --git a/erts/emulator/beam/erl_nif_api_funcs.h b/erts/emulator/beam/erl_nif_api_funcs.h
index 9a8f216773..9163ce25eb 100644
--- a/erts/emulator/beam/erl_nif_api_funcs.h
+++ b/erts/emulator/beam/erl_nif_api_funcs.h
@@ -175,6 +175,8 @@ ERL_NIF_API_FUNC_DECL(size_t, enif_binary_to_term, (ErlNifEnv *env, const unsign
ERL_NIF_API_FUNC_DECL(int, enif_port_command, (ErlNifEnv *env, const ErlNifPort* to_port, ErlNifEnv *msg_env, ERL_NIF_TERM msg));
ERL_NIF_API_FUNC_DECL(int,enif_thread_type,(void));
ERL_NIF_API_FUNC_DECL(int,enif_snprintf,(char * buffer, size_t size, const char *format, ...));
+ERL_NIF_API_FUNC_DECL(int,enif_select,(ErlNifEnv* env, ErlNifEvent e, enum ErlNifSelectFlags flags, void* obj, ERL_NIF_TERM ref));
+ERL_NIF_API_FUNC_DECL(ErlNifResourceType*,enif_open_resource_type_x,(ErlNifEnv*, const char* name_str, const ErlNifResourceTypeInit*, ErlNifResourceFlags flags, ErlNifResourceFlags* tried));
/*
** ADD NEW ENTRIES HERE (before this comment) !!!
@@ -332,6 +334,8 @@ ERL_NIF_API_FUNC_DECL(int,enif_snprintf,(char * buffer, size_t size, const char
# define enif_port_command ERL_NIF_API_FUNC_MACRO(enif_port_command)
# define enif_thread_type ERL_NIF_API_FUNC_MACRO(enif_thread_type)
# define enif_snprintf ERL_NIF_API_FUNC_MACRO(enif_snprintf)
+# define enif_select ERL_NIF_API_FUNC_MACRO(enif_select)
+# define enif_open_resource_type_x ERL_NIF_API_FUNC_MACRO(enif_open_resource_type_x)
/*
** ADD NEW ENTRIES HERE (before this comment)
diff --git a/erts/emulator/beam/global.h b/erts/emulator/beam/global.h
index d5ca3b04eb..2158f54a49 100644
--- a/erts/emulator/beam/global.h
+++ b/erts/emulator/beam/global.h
@@ -42,6 +42,7 @@
#include "erl_utils.h"
#include "erl_port.h"
#include "erl_gc.h"
+#include "erl_nif.h"
struct enif_func_t;
@@ -58,9 +59,37 @@ struct enif_environment_t /* ErlNifEnv */
Process *tracee;
int exiting; /* boolean (dirty nifs might return in exiting state) */
};
+struct enif_resource_type_t
+{
+ struct enif_resource_type_t* next; /* list of all resource types */
+ struct enif_resource_type_t* prev;
+ struct erl_module_nif* owner; /* that created this type and thus implements the destructor*/
+ ErlNifResourceDtor* dtor; /* user destructor function */
+ ErlNifResourceStop* stop;
+ erts_refc_t refc; /* num of resources of this type (HOTSPOT warning)
+ +1 for active erl_module_nif */
+ Eterm module;
+ Eterm name;
+};
+typedef struct enif_resource_t
+{
+ struct enif_resource_type_t* type;
+#ifdef DEBUG
+ erts_refc_t nif_refc;
+# ifdef ARCH_32
+ byte align__[4];
+# endif
+#endif
+
+ char data[1];
+}ErlNifResource;
+
+#define DATA_TO_RESOURCE(PTR) ((ErlNifResource*)((char*)(PTR) - offsetof(ErlNifResource,data)))
+
extern void erts_pre_nif(struct enif_environment_t*, Process*,
struct erl_module_nif*, Process* tracee);
extern void erts_post_nif(struct enif_environment_t* env);
+extern void erts_resource_stop(ErlNifResource*, ErlNifEvent, int is_direct_call);
extern Eterm erts_nif_taints(Process* p);
extern void erts_print_nif_taints(fmtfn_t to, void* to_arg);
void erts_unload_nif(struct erl_module_nif* nif);
diff --git a/erts/emulator/sys/common/erl_check_io.c b/erts/emulator/sys/common/erl_check_io.c
index 44a77f3ea5..4fc95624c7 100644
--- a/erts/emulator/sys/common/erl_check_io.c
+++ b/erts/emulator/sys/common/erl_check_io.c
@@ -53,6 +53,8 @@ typedef char EventStateType;
#define ERTS_EV_TYPE_DRV_SEL ((EventStateType) 1) /* driver_select */
#define ERTS_EV_TYPE_DRV_EV ((EventStateType) 2) /* driver_event */
#define ERTS_EV_TYPE_STOP_USE ((EventStateType) 3) /* pending stop_select */
+#define ERTS_EV_TYPE_NIF ((EventStateType) 4) /* enif_select */
+#define ERTS_EV_TYPE_STOP_NIF ((EventStateType) 5) /* pending nif stop */
typedef char EventStateFlags;
#define ERTS_EV_FLAG_USED ((EventStateFlags) 1) /* ERL_DRV_USE has been turned on */
@@ -122,7 +124,11 @@ typedef struct {
#if ERTS_CIO_HAVE_DRV_EVENT
ErtsDrvEventDataState *event; /* ERTS_EV_TYPE_DRV_EV */
#endif
- erts_driver_t* drv_ptr; /* ERTS_EV_TYPE_STOP_USE */
+ ErtsNifSelectDataState *nif; /* ERTS_EV_TYPE_NIF */
+ union {
+ erts_driver_t* drv_ptr; /* ERTS_EV_TYPE_STOP_USE */
+ ErlNifResource* resource; /* ERTS_EV_TYPE_STOP_NIF */
+ }stop;
} driver;
ErtsPollEvents events;
unsigned short remove_cnt; /* number of removed_fd's referring to this fd */
@@ -211,24 +217,35 @@ static ERTS_INLINE void hash_erase_drv_ev_state(ErtsDrvEventState *state)
#endif /* !ERTS_SYS_CONTINOUS_FD_NUMBERS */
static void stale_drv_select(Eterm id, ErtsDrvEventState *state, int mode);
-static void select_steal(ErlDrvPort ix, ErtsDrvEventState *state,
- int mode, int on);
-static void print_select_op(erts_dsprintf_buf_t *dsbufp,
- ErlDrvPort ix, ErtsSysFdType fd, int mode, int on);
+static void drv_select_steal(ErlDrvPort ix, ErtsDrvEventState *state,
+ int mode, int on);
+static void nif_select_steal(ErtsDrvEventState *state, int mode,
+ ErlNifResource* resource, Eterm ref);
+
+static void print_drv_select_op(erts_dsprintf_buf_t *dsbufp,
+ ErlDrvPort ix, ErtsSysFdType fd, int mode, int on);
+static void print_nif_select_op(erts_dsprintf_buf_t*, ErtsSysFdType,
+ int mode, ErlNifResource*, Eterm ref);
+
#ifdef ERTS_SYS_CONTINOUS_FD_NUMBERS
-static void select_large_fd_error(ErlDrvPort, ErtsSysFdType, int, int);
+static void drv_select_large_fd_error(ErlDrvPort, ErtsSysFdType, int, int);
+static void nif_select_large_fd_error(ErtsSysFdType, int, ErlNifResource*,Eterm ref);
#endif
#if ERTS_CIO_HAVE_DRV_EVENT
-static void event_steal(ErlDrvPort ix, ErtsDrvEventState *state,
+static void drv_event_steal(ErlDrvPort ix, ErtsDrvEventState *state,
ErlDrvEventData event_data);
-static void print_event_op(erts_dsprintf_buf_t *dsbufp,
- ErlDrvPort, ErtsSysFdType, ErlDrvEventData);
+static void print_drv_event_op(erts_dsprintf_buf_t *dsbufp,
+ ErlDrvPort, ErtsSysFdType, ErlDrvEventData);
#ifdef ERTS_SYS_CONTINOUS_FD_NUMBERS
static void event_large_fd_error(ErlDrvPort, ErtsSysFdType, ErlDrvEventData);
#endif
#endif
-static void steal_pending_stop_select(erts_dsprintf_buf_t*, ErlDrvPort,
- ErtsDrvEventState*, int mode, int on);
+static void
+steal_pending_stop_use(erts_dsprintf_buf_t*, ErlDrvPort, ErtsDrvEventState*,
+ int mode, int on);
+static void
+steal_pending_stop_nif(erts_dsprintf_buf_t *dsbufp, ErlNifResource*,
+ ErtsDrvEventState *state, int mode, int on);
#ifdef ERTS_SMP
ERTS_SCHED_PREF_QUICK_ALLOC_IMPL(removed_fd, struct removed_fd, 64, ERTS_ALC_T_FD_LIST)
@@ -263,6 +280,18 @@ alloc_drv_select_data(void)
return dsp;
}
+static ERTS_INLINE ErtsNifSelectDataState *
+alloc_nif_select_data(void)
+{
+ ErtsNifSelectDataState *dsp = erts_alloc(ERTS_ALC_T_NIF_SEL_D_STATE,
+ sizeof(ErtsNifSelectDataState));
+ dsp->in.pid = NIL;
+ dsp->out.pid = NIL;
+ dsp->in.ddeselect_cnt = 0;
+ dsp->out.ddeselect_cnt = 0;
+ return dsp;
+}
+
static ERTS_INLINE void
free_drv_select_data(ErtsDrvSelectDataState *dsp)
{
@@ -271,6 +300,12 @@ free_drv_select_data(ErtsDrvSelectDataState *dsp)
erts_free(ERTS_ALC_T_DRV_SEL_D_STATE, dsp);
}
+static ERTS_INLINE void
+free_nif_select_data(ErtsNifSelectDataState *dsp)
+{
+ erts_free(ERTS_ALC_T_NIF_SEL_D_STATE, dsp);
+}
+
#if ERTS_CIO_HAVE_DRV_EVENT
static ERTS_INLINE ErtsDrvEventDataState *
@@ -352,6 +387,7 @@ forget_removed(struct pollset_info* psi)
erts_smp_spin_unlock(&psi->removed_list_lock);
while (fdlp) {
+ ErlNifResource* resource = NULL;
erts_driver_t* drv_ptr = NULL;
erts_smp_mtx_t* mtx;
ErtsSysFdType fd;
@@ -372,15 +408,25 @@ forget_removed(struct pollset_info* psi)
ASSERT(state->remove_cnt > 0);
if (--state->remove_cnt == 0) {
switch (state->type) {
+ case ERTS_EV_TYPE_STOP_NIF:
+ /* Now we can call stop */
+ resource = state->driver.stop.resource;
+ state->driver.stop.resource = NULL;
+ ASSERT(resource);
+ state->type = ERTS_EV_TYPE_NONE;
+ state->flags &= ~ERTS_EV_FLAG_USED;
+ goto case_ERTS_EV_TYPE_NONE;
+
case ERTS_EV_TYPE_STOP_USE:
/* Now we can call stop_select */
- drv_ptr = state->driver.drv_ptr;
+ drv_ptr = state->driver.stop.drv_ptr;
ASSERT(drv_ptr);
state->type = ERTS_EV_TYPE_NONE;
state->flags &= ~ERTS_EV_FLAG_USED;
- state->driver.drv_ptr = NULL;
+ state->driver.stop.drv_ptr = NULL;
/* Fall through */
- case ERTS_EV_TYPE_NONE:
+ case ERTS_EV_TYPE_NONE:
+ case_ERTS_EV_TYPE_NONE:
#ifndef ERTS_SYS_CONTINOUS_FD_NUMBERS
hash_erase_drv_ev_state(state);
#endif
@@ -403,6 +449,11 @@ forget_removed(struct pollset_info* psi)
erts_ddll_dereference_driver(drv_ptr->handle);
}
}
+ if (resource) {
+ erts_resource_stop(resource, (ErlNifEvent)fd, 0);
+ enif_release_resource(resource->data);
+ }
+
tofree = fdlp;
fdlp = fdlp->next;
removed_fd_free(tofree);
@@ -440,7 +491,8 @@ grow_drv_ev_state(int min_ix)
#if ERTS_CIO_HAVE_DRV_EVENT
drv_ev_state[i].driver.event = NULL;
#endif
- drv_ev_state[i].driver.drv_ptr = NULL;
+ drv_ev_state[i].driver.stop.drv_ptr = NULL;
+ drv_ev_state[i].driver.nif = NULL;
drv_ev_state[i].events = 0;
drv_ev_state[i].remove_cnt = 0;
drv_ev_state[i].type = ERTS_EV_TYPE_NONE;
@@ -480,6 +532,7 @@ abort_tasks(ErtsDrvEventState *state, int mode)
ERTS_EV_TYPE_DRV_EV);
return;
#endif
+ case ERTS_EV_TYPE_NIF:
case ERTS_EV_TYPE_NONE:
return;
default:
@@ -534,6 +587,14 @@ deselect(ErtsDrvEventState *state, int mode)
if (!(state->events)) {
switch (state->type) {
+ case ERTS_EV_TYPE_NIF:
+ state->driver.nif->in.pid = NIL;
+ state->driver.nif->out.pid = NIL;
+ state->driver.nif->in.ddeselect_cnt = 0;
+ state->driver.nif->out.ddeselect_cnt = 0;
+ enif_release_resource(state->driver.stop.resource);
+ state->driver.stop.resource = NULL;
+ break;
case ERTS_EV_TYPE_DRV_SEL:
state->driver.select->inport = NIL;
state->driver.select->outport = NIL;
@@ -569,7 +630,8 @@ check_fd_cleanup(ErtsDrvEventState *state,
#if ERTS_CIO_HAVE_DRV_EVENT
ErtsDrvEventDataState **free_event,
#endif
- ErtsDrvSelectDataState **free_select)
+ ErtsDrvSelectDataState **free_select,
+ ErtsNifSelectDataState **free_nif)
{
erts_aint_t current_cio_time;
@@ -586,6 +648,12 @@ check_fd_cleanup(ErtsDrvEventState *state,
state->driver.select = NULL;
}
+ *free_nif = NULL;
+ if (state->driver.nif && (state->type != ERTS_EV_TYPE_NIF)) {
+ *free_nif = state->driver.nif;
+ state->driver.nif = NULL;
+ }
+
#if ERTS_CIO_HAVE_DRV_EVENT
*free_event = NULL;
if (state->driver.event
@@ -617,12 +685,14 @@ check_cleanup_active_fd(ErtsSysFdType fd,
ErtsPollControlEntry *pce,
int *pce_ix,
#endif
- erts_aint_t current_cio_time)
+ erts_aint_t current_cio_time,
+ int may_sleep)
{
ErtsDrvEventState *state;
int active = 0;
erts_smp_mtx_t *mtx = fd_mtx(fd);
void *free_select = NULL;
+ void *free_nif = NULL;
#if ERTS_CIO_HAVE_DRV_EVENT
void *free_event = NULL;
#endif
@@ -682,6 +752,41 @@ check_cleanup_active_fd(ErtsSysFdType fd,
}
}
+ if (state->driver.nif) {
+#if ERTS_CIO_DEFER_ACTIVE_EVENTS
+# error Windows
+#endif
+ int do_wake = 0;
+ ErtsPollEvents rm_events = 0;
+ if (state->driver.nif->in.ddeselect_cnt) {
+ ASSERT(state->type == ERTS_EV_TYPE_NIF);
+ ASSERT(state->events & ERTS_POLL_EV_IN);
+ ASSERT(is_nil(state->driver.nif->in.pid));
+ if (may_sleep || state->driver.nif->in.ddeselect_cnt == 1) {
+ rm_events = ERTS_POLL_EV_IN;
+ state->driver.nif->in.ddeselect_cnt = 0;
+ }
+ }
+ if (state->driver.nif->out.ddeselect_cnt) {
+ ASSERT(state->type == ERTS_EV_TYPE_NIF);
+ ASSERT(state->events & ERTS_POLL_EV_OUT);
+ ASSERT(is_nil(state->driver.nif->out.pid));
+ if (may_sleep || state->driver.nif->out.ddeselect_cnt == 1) {
+ rm_events |= ERTS_POLL_EV_OUT;
+ state->driver.nif->out.ddeselect_cnt = 0;
+ }
+ }
+ if (rm_events) {
+ state->events = ERTS_CIO_POLL_CTL(pollset.ps, state->fd, rm_events, 0, &do_wake);
+ }
+ if (state->events)
+ active = 1;
+ else if (state->type != ERTS_EV_TYPE_NIF) {
+ free_nif = state->driver.nif;
+ state->driver.nif = NULL;
+ }
+ }
+
#if ERTS_CIO_HAVE_DRV_EVENT
if (state->driver.event) {
if (is_iotask_active(&state->driver.event->iotask, current_cio_time)) {
@@ -722,6 +827,8 @@ check_cleanup_active_fd(ErtsSysFdType fd,
if (free_select)
free_drv_select_data(free_select);
+ if (free_nif)
+ free_nif_select_data(free_nif);
#if ERTS_CIO_HAVE_DRV_EVENT
if (free_event)
free_drv_event_data(free_event);
@@ -746,7 +853,7 @@ check_cleanup_active_fd(ErtsSysFdType fd,
}
static void
-check_cleanup_active_fds(erts_aint_t current_cio_time)
+check_cleanup_active_fds(erts_aint_t current_cio_time, int may_sleep)
{
int six = pollset.active_fd.six;
int eix = pollset.active_fd.eix;
@@ -773,7 +880,8 @@ check_cleanup_active_fds(erts_aint_t current_cio_time)
pctrl_entries,
&pctrl_ix,
#endif
- current_cio_time)) {
+ current_cio_time,
+ may_sleep)) {
no--;
if (ix == six) {
#ifdef DEBUG
@@ -807,13 +915,30 @@ check_cleanup_active_fds(erts_aint_t current_cio_time)
erts_smp_atomic32_set_relb(&pollset.active_fd.no, no);
}
+static void grow_active_fds(void)
+{
+ ASSERT(pollset.active_fd.six == pollset.active_fd.eix);
+ pollset.active_fd.six = 0;
+ pollset.active_fd.eix = pollset.active_fd.size;
+ pollset.active_fd.size += ERTS_ACTIVE_FD_INC;
+ pollset.active_fd.array = erts_realloc(ERTS_ALC_T_ACTIVE_FD_ARR,
+ pollset.active_fd.array,
+ pollset.active_fd.size*sizeof(ErtsSysFdType));
+#ifdef DEBUG
+ {
+ int i;
+ for (i = pollset.active_fd.eix + 1; i < pollset.active_fd.size; i++)
+ pollset.active_fd.array[i] = ERTS_SYS_FD_INVALID;
+ }
+#endif
+}
+
static ERTS_INLINE void
add_active_fd(ErtsSysFdType fd)
{
int eix = pollset.active_fd.eix;
int size = pollset.active_fd.size;
-
pollset.active_fd.array[eix] = fd;
erts_smp_atomic32_set_relb(&pollset.active_fd.no,
@@ -823,25 +948,11 @@ add_active_fd(ErtsSysFdType fd)
eix++;
if (eix >= size)
eix = 0;
- if (pollset.active_fd.six == eix) {
- pollset.active_fd.six = 0;
- eix = size;
- size += ERTS_ACTIVE_FD_INC;
- pollset.active_fd.array = erts_realloc(ERTS_ALC_T_ACTIVE_FD_ARR,
- pollset.active_fd.array,
- sizeof(ErtsSysFdType)*size);
- pollset.active_fd.size = size;
-#ifdef DEBUG
- {
- int i;
- for (i = eix + 1; i < size; i++)
- pollset.active_fd.array[i] = ERTS_SYS_FD_INVALID;
- }
-#endif
+ pollset.active_fd.eix = eix;
+ if (pollset.active_fd.six == eix) {
+ grow_active_fds();
}
-
- pollset.active_fd.eix = eix;
}
int
@@ -863,6 +974,7 @@ ERTS_CIO_EXPORT(driver_select)(ErlDrvPort ix,
ErtsDrvEventDataState *free_event = NULL;
#endif
ErtsDrvSelectDataState *free_select = NULL;
+ ErtsNifSelectDataState *free_nif = NULL;
#ifdef USE_VM_PROBES
DTRACE_CHARBUF(name, 64);
#endif
@@ -878,7 +990,7 @@ ERTS_CIO_EXPORT(driver_select)(ErlDrvPort ix,
return -1;
}
if (fd >= max_fds) {
- select_large_fd_error(ix, fd, mode, on);
+ drv_select_large_fd_error(ix, fd, mode, on);
return -1;
}
grow_drv_ev_state(fd);
@@ -916,26 +1028,38 @@ ERTS_CIO_EXPORT(driver_select)(ErlDrvPort ix,
}
#endif
+ switch (state->type) {
#if ERTS_CIO_HAVE_DRV_EVENT
- if (state->type == ERTS_EV_TYPE_DRV_EV)
- select_steal(ix, state, mode, on);
+ case ERTS_EV_TYPE_DRV_EV:
#endif
- if (state->type == ERTS_EV_TYPE_STOP_USE) {
- erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
- print_select_op(dsbufp, ix, state->fd, mode, on);
- steal_pending_stop_select(dsbufp, ix, state, mode, on);
- if (state->type == ERTS_EV_TYPE_STOP_USE) {
- ret = 0;
- goto done; /* stop_select still pending */
- }
- ASSERT(state->type == ERTS_EV_TYPE_NONE);
- }
+ case ERTS_EV_TYPE_NIF:
+ drv_select_steal(ix, state, mode, on);
+ break;
+ case ERTS_EV_TYPE_STOP_USE: {
+ erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
+ print_drv_select_op(dsbufp, ix, state->fd, mode, on);
+ steal_pending_stop_use(dsbufp, ix, state, mode, on);
+ if (state->type == ERTS_EV_TYPE_STOP_USE) {
+ ret = 0;
+ goto done; /* stop_select still pending */
+ }
+ ASSERT(state->type == ERTS_EV_TYPE_NONE);
+ break;
+ }
+ case ERTS_EV_TYPE_STOP_NIF: {
+ erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
+ print_drv_select_op(dsbufp, ix, state->fd, mode, on);
+ steal_pending_stop_nif(dsbufp, NULL, state, mode, on);
+ ASSERT(state->type == ERTS_EV_TYPE_NONE);
+ break;
+
+ }}
if (mode & ERL_DRV_READ) {
if (state->type == ERTS_EV_TYPE_DRV_SEL) {
Eterm owner = state->driver.select->inport;
if (owner != id && is_not_nil(owner))
- select_steal(ix, state, mode, on);
+ drv_select_steal(ix, state, mode, on);
}
ctl_events |= ERTS_POLL_EV_IN;
}
@@ -943,7 +1067,7 @@ ERTS_CIO_EXPORT(driver_select)(ErlDrvPort ix,
if (state->type == ERTS_EV_TYPE_DRV_SEL) {
Eterm owner = state->driver.select->outport;
if (owner != id && is_not_nil(owner))
- select_steal(ix, state, mode, on);
+ drv_select_steal(ix, state, mode, on);
}
ctl_events |= ERTS_POLL_EV_OUT;
}
@@ -1033,7 +1157,7 @@ ERTS_CIO_EXPORT(driver_select)(ErlDrvPort ix,
else {
/* Not safe to close fd, postpone stop_select callback. */
state->type = ERTS_EV_TYPE_STOP_USE;
- state->driver.drv_ptr = drv_ptr;
+ state->driver.stop.drv_ptr = drv_ptr;
if (drv_ptr->handle) {
erts_ddll_reference_referenced_driver(drv_ptr->handle);
}
@@ -1050,7 +1174,8 @@ done:
#if ERTS_CIO_HAVE_DRV_EVENT
&free_event,
#endif
- &free_select);
+ &free_select,
+ &free_nif);
done_unknown:
erts_smp_mtx_unlock(fd_mtx(fd));
@@ -1063,6 +1188,263 @@ done_unknown:
}
if (free_select)
free_drv_select_data(free_select);
+ if (free_nif)
+ free_nif_select_data(free_nif);
+
+#if ERTS_CIO_HAVE_DRV_EVENT
+ if (free_event)
+ free_drv_event_data(free_event);
+#endif
+ return ret;
+}
+
+enum ErlNifSelectReturn
+ERTS_CIO_EXPORT(enif_select)(ErlNifEnv* env,
+ ErlNifEvent e,
+ enum ErlNifSelectFlags mode,
+ void* obj,
+ Eterm ref)
+{
+ int on;
+ const Eterm id = env->proc->common.id;
+ ErlNifResource* resource = DATA_TO_RESOURCE(obj);
+ ErtsSysFdType fd = (ErtsSysFdType) e;
+ ErtsPollEvents ctl_events = (ErtsPollEvents) 0;
+ ErtsPollEvents new_events, old_events;
+ ErtsDrvEventState *state;
+ int wake_poller;
+ enum ErlNifSelectReturn ret;
+ enum { NO_STOP=0, CALL_STOP, CALL_STOP_AND_RELEASE } call_stop = NO_STOP;
+#if ERTS_CIO_HAVE_DRV_EVENT
+ ErtsDrvEventDataState *free_event = NULL;
+#endif
+ ErtsDrvSelectDataState *free_select = NULL;
+ ErtsNifSelectDataState *free_nif = NULL;
+#ifdef USE_VM_PROBES
+ DTRACE_CHARBUF(name, 64);
+#endif
+
+#ifdef ERTS_SYS_CONTINOUS_FD_NUMBERS
+ if ((unsigned)fd >= (unsigned)erts_smp_atomic_read_nob(&drv_ev_state_len)) {
+ if (fd < 0) {
+ return ERL_NIF_SELECT_ERROR | ERL_NIF_SELECT_INVALID_EVENT;
+ }
+ if (fd >= max_fds) {
+ nif_select_large_fd_error(fd, mode, resource, ref);
+ return ERL_NIF_SELECT_ERROR | ERL_NIF_SELECT_INVALID_EVENT;
+ }
+ grow_drv_ev_state(fd);
+ }
+#endif
+
+ erts_smp_mtx_lock(fd_mtx(fd));
+
+#ifdef ERTS_SYS_CONTINOUS_FD_NUMBERS
+ state = &drv_ev_state[(int) fd];
+#else
+ state = hash_get_drv_ev_state(fd); /* may be NULL! */
+#endif
+
+ if (mode & ERL_NIF_SELECT_STOP) {
+ ASSERT(resource->type->stop);
+ if (IS_FD_UNKNOWN(state)) {
+ /* fast track to stop callback */
+ call_stop = CALL_STOP;
+ ret = ERL_NIF_SELECT_STOP_CALLED;
+ goto done_unknown;
+ }
+ on = 0;
+ mode = ERL_DRV_READ | ERL_DRV_WRITE | ERL_DRV_USE;
+ wake_poller = 1; /* to eject fd from pollset (if needed) */
+ ctl_events = ERTS_POLL_EV_IN | ERTS_POLL_EV_OUT;
+ }
+ else {
+ on = 1;
+ ASSERT(mode);
+ wake_poller = 0;
+ if (mode & ERL_DRV_READ) {
+ ctl_events |= ERTS_POLL_EV_IN;
+ }
+ if (mode & ERL_DRV_WRITE) {
+ ctl_events |= ERTS_POLL_EV_OUT;
+ }
+ }
+
+#ifndef ERTS_SYS_CONTINOUS_FD_NUMBERS
+ if (state == NULL) {
+ state = hash_new_drv_ev_state(fd);
+ }
+#endif
+
+ switch (state->type) {
+ case ERTS_EV_TYPE_NIF:
+ /*
+ * Changing resource is considered stealing.
+ * Changing process and/or ref is ok (I think?).
+ */
+ if (state->driver.stop.resource != resource)
+ nif_select_steal(state, ERL_DRV_READ | ERL_DRV_WRITE, resource, ref);
+ break;
+#if ERTS_CIO_HAVE_DRV_EVENT
+ case ERTS_EV_TYPE_DRV_EV:
+#endif
+ case ERTS_EV_TYPE_DRV_SEL:
+ nif_select_steal(state, mode, resource, ref);
+ break;
+ case ERTS_EV_TYPE_STOP_USE: {
+ erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
+ print_nif_select_op(dsbufp, fd, mode, resource, ref);
+ steal_pending_stop_use(dsbufp, ERTS_INVALID_ERL_DRV_PORT, state, mode, on);
+ ASSERT(state->type == ERTS_EV_TYPE_NONE);
+ break;
+ }
+ case ERTS_EV_TYPE_STOP_NIF: {
+ erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
+ print_nif_select_op(dsbufp, fd, mode, resource, ref);
+ steal_pending_stop_nif(dsbufp, resource, state, mode, on);
+ if (state->type == ERTS_EV_TYPE_STOP_NIF) {
+ ret = ERL_NIF_SELECT_STOP_SCHEDULED; /* ?? */
+ goto done;
+ }
+ ASSERT(state->type == ERTS_EV_TYPE_NONE);
+ break;
+ }}
+
+ ASSERT((state->type == ERTS_EV_TYPE_NIF) ||
+ (state->type == ERTS_EV_TYPE_NONE && !state->events));
+
+ new_events = ERTS_CIO_POLL_CTL(pollset.ps, state->fd, ctl_events, on, &wake_poller);
+
+ if (new_events & (ERTS_POLL_EV_ERR|ERTS_POLL_EV_NVAL)) {
+ if (state->type == ERTS_EV_TYPE_NIF && !state->events) {
+ state->type = ERTS_EV_TYPE_NONE;
+ state->flags &= ~ERTS_EV_FLAG_USED;
+ state->driver.nif->in.pid = NIL;
+ state->driver.nif->out.pid = NIL;
+ state->driver.nif->in.ddeselect_cnt = 0;
+ state->driver.nif->out.ddeselect_cnt = 0;
+ state->driver.stop.resource = NULL;
+ }
+ ret = ERL_NIF_SELECT_ERROR | ERL_NIF_SELECT_FAILED;
+ goto done;
+ }
+
+ old_events = state->events;
+
+ ASSERT(on
+ ? (new_events == (state->events | ctl_events))
+ : (new_events == (state->events & ~ctl_events)));
+
+ ASSERT(state->type == ERTS_EV_TYPE_NIF
+ || state->type == ERTS_EV_TYPE_NONE);
+
+ state->events = new_events;
+ if (on) {
+ Uint32* refn;
+ if (!state->driver.nif)
+ state->driver.nif = alloc_nif_select_data();
+ if (state->type == ERTS_EV_TYPE_NONE) {
+ state->type = ERTS_EV_TYPE_NIF;
+ state->driver.stop.resource = resource;
+ enif_keep_resource(resource->data);
+ }
+ ASSERT(state->type == ERTS_EV_TYPE_NIF);
+ ASSERT(state->driver.stop.resource == resource);
+ if (ctl_events & ERTS_POLL_EV_IN) {
+ state->driver.nif->in.pid = id;
+ if (is_immed(ref)) {
+ state->driver.nif->in.immed = ref;
+ } else {
+ ASSERT(is_internal_ref(ref));
+ refn = internal_ref_numbers(ref);
+ state->driver.nif->in.immed = THE_NON_VALUE;
+ state->driver.nif->in.refn[0] = refn[0];
+ state->driver.nif->in.refn[1] = refn[1];
+ state->driver.nif->in.refn[2] = refn[2];
+ }
+ state->driver.nif->in.ddeselect_cnt = 0;
+ }
+ if (ctl_events & ERTS_POLL_EV_OUT) {
+ state->driver.nif->out.pid = id;
+ if (is_immed(ref)) {
+ state->driver.nif->out.immed = ref;
+ } else {
+ ASSERT(is_internal_ref(ref));
+ refn = internal_ref_numbers(ref);
+ state->driver.nif->out.immed = THE_NON_VALUE;
+ state->driver.nif->out.refn[0] = refn[0];
+ state->driver.nif->out.refn[1] = refn[1];
+ state->driver.nif->out.refn[2] = refn[2];
+ }
+ state->driver.nif->out.ddeselect_cnt = 0;
+ }
+ ret = 0;
+ }
+ else { /* off */
+ if (state->type == ERTS_EV_TYPE_NIF) {
+ //erts_fprintf(stderr, "SVERK: enif select clear fd=%d inpid=%T inrsrc=%p\n",
+ // state->fd, state->driver.nif->inpid,
+ // state->driver.nif->in.resource);
+ state->driver.nif->in.pid = NIL;
+ state->driver.nif->out.pid = NIL;
+ state->driver.nif->in.ddeselect_cnt = 0;
+ state->driver.nif->out.ddeselect_cnt = 0;
+ if (old_events != 0) {
+ remember_removed(state, &pollset);
+ }
+ }
+ ASSERT(new_events==0);
+ if (state->remove_cnt == 0 || !wake_poller) {
+ /* Safe to close fd now as it is not in pollset
+ or there was no need to eject fd (kernel poll) */
+ //erts_fprintf(stderr, "SVERK : enif_select calling stop before return\n");
+ if (state->type == ERTS_EV_TYPE_NIF) {
+ ASSERT(state->driver.stop.resource == resource);
+ call_stop = CALL_STOP_AND_RELEASE;
+ state->driver.stop.resource = NULL;
+ }
+ else {
+ ASSERT(!state->driver.stop.resource);
+ call_stop = CALL_STOP;
+ }
+ state->type = ERTS_EV_TYPE_NONE;
+ ret = ERL_NIF_SELECT_STOP_CALLED;
+ }
+ else {
+ /* Not safe to close fd, postpone stop_select callback. */
+ //erts_fprintf(stderr, "SVERK: enif_select schedule stop\n");
+ if (state->type == ERTS_EV_TYPE_NONE) {
+ ASSERT(!state->driver.stop.resource);
+ state->driver.stop.resource = resource;
+ enif_keep_resource(resource);
+ }
+ state->type = ERTS_EV_TYPE_STOP_NIF;
+ ret = ERL_NIF_SELECT_STOP_SCHEDULED;
+ }
+ }
+
+done:
+
+ check_fd_cleanup(state,
+#if ERTS_CIO_HAVE_DRV_EVENT
+ &free_event,
+#endif
+ &free_select,
+ &free_nif);
+
+done_unknown:
+ erts_smp_mtx_unlock(fd_mtx(fd));
+ if (call_stop) {
+ erts_resource_stop(resource, (ErlNifEvent)fd, 1);
+ if (call_stop == CALL_STOP_AND_RELEASE) {
+ enif_release_resource(resource->data);
+ }
+ }
+ if (free_select)
+ free_drv_select_data(free_select);
+ if (free_nif)
+ free_nif_select_data(free_nif);
+
#if ERTS_CIO_HAVE_DRV_EVENT
if (free_event)
free_drv_event_data(free_event);
@@ -1070,6 +1452,7 @@ done_unknown:
return ret;
}
+
int
ERTS_CIO_EXPORT(driver_event)(ErlDrvPort ix,
ErlDrvEvent e,
@@ -1090,6 +1473,7 @@ ERTS_CIO_EXPORT(driver_event)(ErlDrvPort ix,
ErtsDrvEventDataState *free_event;
#endif
ErtsDrvSelectDataState *free_select;
+ ErtsNifSelectDataState *free_nif;
Port *prt = erts_drvport2port(ix);
if (prt == ERTS_INVALID_ERL_DRV_PORT)
@@ -1126,12 +1510,12 @@ ERTS_CIO_EXPORT(driver_event)(ErlDrvPort ix,
if (state->driver.event->port == id) break;
/*fall through*/
case ERTS_EV_TYPE_DRV_SEL:
- event_steal(ix, state, event_data);
+ drv_event_steal(ix, state, event_data);
break;
case ERTS_EV_TYPE_STOP_USE: {
erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
- print_event_op(dsbufp, ix, fd, event_data);
- steal_pending_stop_select(dsbufp, ix, state, 0, 1);
+ print_drv_event_op(dsbufp, ix, fd, event_data);
+ steal_pending_stop_use(dsbufp, ix, state, 0, 1);
break;
}
}
@@ -1199,12 +1583,15 @@ done:
#if ERTS_CIO_HAVE_DRV_EVENT
&free_event,
#endif
- &free_select);
+ &free_select,
+ &free_nif);
erts_smp_mtx_unlock(fd_mtx(fd));
if (free_select)
free_drv_select_data(free_select);
+ if (free_nif)
+ free_nif_select_data(free_nif);
#if ERTS_CIO_HAVE_DRV_EVENT
if (free_event)
free_drv_event_data(free_event);
@@ -1240,13 +1627,19 @@ need2steal(ErtsDrvEventState *state, int mode)
state,
ERL_DRV_WRITE);
break;
+ case ERTS_EV_TYPE_NIF:
+ ASSERT(state->driver.stop.resource);
+ do_steal = 1;
+ break;
+
#if ERTS_CIO_HAVE_DRV_EVENT
case ERTS_EV_TYPE_DRV_EV:
do_steal |= chk_stale(state->driver.event->port, state, 0);
break;
#endif
case ERTS_EV_TYPE_STOP_USE:
- ASSERT(0);
+ case ERTS_EV_TYPE_STOP_NIF:
+ ASSERT(0);
break;
default:
break;
@@ -1307,6 +1700,25 @@ steal(erts_dsprintf_buf_t *dsbufp, ErtsDrvEventState *state, int mode)
erts_dsprintf(dsbufp, "\n");
break;
}
+ case ERTS_EV_TYPE_NIF: {
+ Eterm iid = state->driver.nif->in.pid;
+ Eterm oid = state->driver.nif->out.pid;
+ const char* with = "with";
+ ErlNifResourceType* rt = state->driver.stop.resource->type;
+
+ erts_dsprintf(dsbufp, "resource %T:%T", rt->module, rt->name);
+
+ if (is_not_nil(iid)) {
+ erts_dsprintf(dsbufp, " %s in-pid %T", with, iid);
+ with = "and";
+ }
+ if (is_not_nil(oid)) {
+ erts_dsprintf(dsbufp, " %s out-pid %T", with, oid);
+ }
+ deselect(state, 0);
+ erts_dsprintf(dsbufp, "\n");
+ break;
+ }
#if ERTS_CIO_HAVE_DRV_EVENT
case ERTS_EV_TYPE_DRV_EV: {
Eterm eid = state->driver.event->port;
@@ -1324,7 +1736,8 @@ steal(erts_dsprintf_buf_t *dsbufp, ErtsDrvEventState *state, int mode)
break;
}
#endif
- case ERTS_EV_TYPE_STOP_USE: {
+ case ERTS_EV_TYPE_STOP_USE:
+ case ERTS_EV_TYPE_STOP_NIF: {
ASSERT(0);
break;
}
@@ -1335,8 +1748,8 @@ steal(erts_dsprintf_buf_t *dsbufp, ErtsDrvEventState *state, int mode)
}
static void
-print_select_op(erts_dsprintf_buf_t *dsbufp,
- ErlDrvPort ix, ErtsSysFdType fd, int mode, int on)
+print_drv_select_op(erts_dsprintf_buf_t *dsbufp,
+ ErlDrvPort ix, ErtsSysFdType fd, int mode, int on)
{
Port *pp = erts_drvport2port(ix);
erts_dsprintf(dsbufp,
@@ -1354,11 +1767,40 @@ print_select_op(erts_dsprintf_buf_t *dsbufp,
}
static void
-select_steal(ErlDrvPort ix, ErtsDrvEventState *state, int mode, int on)
+print_nif_select_op(erts_dsprintf_buf_t *dsbufp,
+ ErtsSysFdType fd, int mode,
+ ErlNifResource* resource, Eterm ref)
+{
+ erts_dsprintf(dsbufp,
+ "enif_select(_, %d,%s%s%s, %T:%T, %T) ",
+ (int) GET_FD(fd),
+ mode & ERL_NIF_SELECT_READ ? " READ" : "",
+ mode & ERL_NIF_SELECT_WRITE ? " WRITE" : "",
+ mode & ERL_NIF_SELECT_STOP ? " STOP" : "",
+ resource->type->module,
+ resource->type->name,
+ ref);
+}
+
+
+static void
+drv_select_steal(ErlDrvPort ix, ErtsDrvEventState *state, int mode, int on)
{
if (need2steal(state, mode)) {
erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
- print_select_op(dsbufp, ix, state->fd, mode, on);
+ print_drv_select_op(dsbufp, ix, state->fd, mode, on);
+ steal(dsbufp, state, mode);
+ erts_send_error_to_logger_nogl(dsbufp);
+ }
+}
+
+static void
+nif_select_steal(ErtsDrvEventState *state, int mode,
+ ErlNifResource* resource, Eterm ref)
+{
+ if (need2steal(state, mode)) {
+ erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
+ print_nif_select_op(dsbufp, state->fd, mode, resource, ref);
steal(dsbufp, state, mode);
erts_send_error_to_logger_nogl(dsbufp);
}
@@ -1374,10 +1816,20 @@ large_fd_error_common(erts_dsprintf_buf_t *dsbufp, ErtsSysFdType fd)
}
static void
-select_large_fd_error(ErlDrvPort ix, ErtsSysFdType fd, int mode, int on)
+drv_select_large_fd_error(ErlDrvPort ix, ErtsSysFdType fd, int mode, int on)
+{
+ erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
+ print_drv_select_op(dsbufp, ix, fd, mode, on);
+ erts_dsprintf(dsbufp, "failed: ");
+ large_fd_error_common(dsbufp, fd);
+ erts_send_error_to_logger_nogl(dsbufp);
+}
+static void
+nif_select_large_fd_error(ErtsSysFdType fd, int mode,
+ ErlNifResource* resource, Eterm ref)
{
erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
- print_select_op(dsbufp, ix, fd, mode, on);
+ print_nif_select_op(dsbufp, fd, mode, resource, ref);
erts_dsprintf(dsbufp, "failed: ");
large_fd_error_common(dsbufp, fd);
erts_send_error_to_logger_nogl(dsbufp);
@@ -1387,41 +1839,81 @@ select_large_fd_error(ErlDrvPort ix, ErtsSysFdType fd, int mode, int on)
static void
-steal_pending_stop_select(erts_dsprintf_buf_t *dsbufp, ErlDrvPort ix,
- ErtsDrvEventState *state, int mode, int on)
+steal_pending_stop_use(erts_dsprintf_buf_t *dsbufp, ErlDrvPort ix,
+ ErtsDrvEventState *state, int mode, int on)
{
+ int cancel = 0;
ASSERT(state->type == ERTS_EV_TYPE_STOP_USE);
- erts_dsprintf(dsbufp, "failed: fd=%d (re)selected before stop_select "
- "was called for driver %s\n",
- (int) GET_FD(state->fd), state->driver.drv_ptr->name);
- erts_send_error_to_logger_nogl(dsbufp);
if (on) {
/* Either fd-owner changed its mind about closing
* or closed fd before stop_select callback and fd is now reused.
* In either case stop_select should not be called.
- */
- state->type = ERTS_EV_TYPE_NONE;
- state->flags &= ~ERTS_EV_FLAG_USED;
- if (state->driver.drv_ptr->handle) {
- erts_ddll_dereference_driver(state->driver.drv_ptr->handle);
- }
- state->driver.drv_ptr = NULL;
+ */
+ cancel = 1;
}
else if ((mode & ERL_DRV_USE_NO_CALLBACK) == ERL_DRV_USE) {
Port *prt = erts_drvport2port(ix);
- erts_driver_t* drv_ptr = prt != ERTS_INVALID_ERL_DRV_PORT ? prt->drv_ptr : NULL;
- if (drv_ptr && drv_ptr != state->driver.drv_ptr) {
- /* Some other driver wants the stop_select callback */
- if (state->driver.drv_ptr->handle) {
- erts_ddll_dereference_driver(state->driver.drv_ptr->handle);
- }
- if (drv_ptr->handle) {
- erts_ddll_reference_referenced_driver(drv_ptr->handle);
- }
- state->driver.drv_ptr = drv_ptr;
- }
+ if (prt == ERTS_INVALID_ERL_DRV_PORT
+ || prt->drv_ptr != state->driver.stop.drv_ptr) {
+ /* Some other driver or nif wants the stop_select callback */
+ cancel = 1;
+ }
+ }
+
+ if (cancel) {
+ erts_dsprintf(dsbufp, "called before stop_select was called for driver '%s'\n",
+ state->driver.stop.drv_ptr->name);
+ if (state->driver.stop.drv_ptr->handle) {
+ erts_ddll_dereference_driver(state->driver.stop.drv_ptr->handle);
+ }
+ state->type = ERTS_EV_TYPE_NONE;
+ state->flags &= ~ERTS_EV_FLAG_USED;
+ state->driver.stop.drv_ptr = NULL;
+ }
+ else {
+ erts_dsprintf(dsbufp, "ignored repeated call\n");
}
+ erts_send_error_to_logger_nogl(dsbufp);
+}
+
+static void
+steal_pending_stop_nif(erts_dsprintf_buf_t *dsbufp, ErlNifResource* resource,
+ ErtsDrvEventState *state, int mode, int on)
+{
+ int cancel = 0;
+
+ ASSERT(state->type == ERTS_EV_TYPE_STOP_NIF);
+ ASSERT(state->driver.stop.resource);
+
+ if (on) {
+ ASSERT(mode & (ERL_NIF_SELECT_READ | ERL_NIF_SELECT_WRITE));
+ /* Either fd-owner changed its mind about closing
+ * or closed fd before stop callback and fd is now reused.
+ * In either case, stop should not be called.
+ */
+ cancel = 1;
+ }
+ else if ((mode & ERL_DRV_USE_NO_CALLBACK) == ERL_DRV_USE
+ && resource != state->driver.stop.resource) {
+ /* Some driver or other resource wants the stop callback */
+ cancel = 1;
+ }
+
+ if (cancel) {
+ ErlNifResourceType* rt = state->driver.stop.resource->type;
+ erts_dsprintf(dsbufp, "called before stop was called for NIF resource %T:%T\n",
+ rt->module, rt->name);
+
+ enif_release_resource(state->driver.stop.resource);
+ state->type = ERTS_EV_TYPE_NONE;
+ state->flags &= ~ERTS_EV_FLAG_USED;
+ state->driver.stop.resource = NULL;
+ }
+ else {
+ erts_dsprintf(dsbufp, "ignored repeated call\n");
+ }
+ erts_send_error_to_logger_nogl(dsbufp);
}
@@ -1429,8 +1921,8 @@ steal_pending_stop_select(erts_dsprintf_buf_t *dsbufp, ErlDrvPort ix,
#if ERTS_CIO_HAVE_DRV_EVENT
static void
-print_event_op(erts_dsprintf_buf_t *dsbufp,
- ErlDrvPort ix, ErtsSysFdType fd, ErlDrvEventData event_data)
+print_drv_event_op(erts_dsprintf_buf_t *dsbufp,
+ ErlDrvPort ix, ErtsSysFdType fd, ErlDrvEventData event_data)
{
Port *pp = erts_drvport2port(ix);
erts_dsprintf(dsbufp, "driver_event(%p, %d, ", ix, (int) fd);
@@ -1447,11 +1939,11 @@ print_event_op(erts_dsprintf_buf_t *dsbufp,
}
static void
-event_steal(ErlDrvPort ix, ErtsDrvEventState *state, ErlDrvEventData event_data)
+drv_event_steal(ErlDrvPort ix, ErtsDrvEventState *state, ErlDrvEventData event_data)
{
if (need2steal(state, ERL_DRV_READ|ERL_DRV_WRITE)) {
erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
- print_event_op(dsbufp, ix, state->fd, event_data);
+ print_drv_event_op(dsbufp, ix, state->fd, event_data);
steal(dsbufp, state, ERL_DRV_READ|ERL_DRV_WRITE);
erts_send_error_to_logger_nogl(dsbufp);
}
@@ -1466,7 +1958,7 @@ static void
event_large_fd_error(ErlDrvPort ix, ErtsSysFdType fd, ErlDrvEventData event_data)
{
erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
- print_event_op(dsbufp, ix, fd, event_data);
+ print_drv_event_op(dsbufp, ix, fd, event_data);
erts_dsprintf(dsbufp, "failed: ");
large_fd_error_common(dsbufp, fd);
erts_send_error_to_logger_nogl(dsbufp);
@@ -1553,6 +2045,56 @@ oready(Eterm id, ErtsDrvEventState *state, erts_aint_t current_cio_time)
}
}
+static ERTS_INLINE void
+send_event_tuple(struct erts_nif_select_event* e, ErlNifResource* resource,
+ Eterm event_atom)
+{
+ Process* rp = erts_proc_lookup(e->pid);
+ ErtsProcLocks rp_locks = 0;
+ ErtsMessage* mp;
+ ErlOffHeap* ohp;
+ ErtsBinary* bin;
+ Eterm* hp;
+ Uint hsz;
+ Eterm resource_term, ref_term, tuple;
+
+ if (!rp) {
+ erts_fprintf(stderr, "SVERK: Process %T not alive for msg %T\n", e->pid, event_atom);
+ return;
+ }
+
+ bin = ERTS_MAGIC_BIN_FROM_UNALIGNED_DATA(resource);
+
+ /* {select, Resource, Ref, EventAtom} */
+ if (is_value(e->immed)) {
+ hsz = 5 + PROC_BIN_SIZE;
+ }
+ else {
+ hsz = 5 + PROC_BIN_SIZE + REF_THING_SIZE;
+ }
+
+ mp = erts_alloc_message_heap(rp, &rp_locks, hsz, &hp, &ohp);
+
+ resource_term = erts_mk_magic_binary_term(&hp, ohp, &bin->binary);
+ if (is_value(e->immed)) {
+ ASSERT(is_immed(e->immed));
+ ref_term = e->immed;
+ }
+ else {
+ write_ref_thing(hp, e->refn[0], e->refn[1], e->refn[2]);
+ ref_term = make_internal_ref(hp);
+ hp += REF_THING_SIZE;
+ }
+ tuple = TUPLE4(hp, am_select, resource_term, ref_term, event_atom);
+
+ ERL_MESSAGE_TOKEN(mp) = am_undefined;
+ erts_queue_message(rp, rp_locks, mp, tuple, am_system);
+
+ if (rp_locks)
+ erts_smp_proc_unlock(rp, rp_locks);
+}
+
+
#if ERTS_CIO_HAVE_DRV_EVENT
static ERTS_INLINE void
eready(Eterm id, ErtsDrvEventState *state, ErlDrvEventData event_data,
@@ -1598,6 +2140,12 @@ ERTS_CIO_EXPORT(erts_check_io_interrupt_timed)(int set,
ERTS_CIO_POLL_INTR_TMD(pollset.ps, set, timeout_time);
}
+/*
+ * Number of ignored events, for a lingering fd added by enif_select(),
+ * until we deselect fd-event from pollset.
+ */
+#define ERTS_NIF_DELAYED_DESELECT 20
+
void
ERTS_CIO_EXPORT(erts_check_io)(int do_wait)
{
@@ -1631,7 +2179,8 @@ ERTS_CIO_EXPORT(erts_check_io)(int do_wait)
current_cio_time++;
erts_smp_atomic_set_relb(&erts_check_io_time, current_cio_time);
- check_cleanup_active_fds(current_cio_time);
+ check_cleanup_active_fds(current_cio_time,
+ timeout_time != ERTS_POLL_NO_TIMEOUT);
#ifdef ERTS_ENABLE_LOCK_CHECK
erts_lc_check_exact(NULL, 0); /* No locks should be locked */
@@ -1699,31 +2248,22 @@ ERTS_CIO_EXPORT(erts_check_io)(int do_wait)
switch (state->type) {
case ERTS_EV_TYPE_DRV_SEL: { /* Requested via driver_select()... */
- ErtsPollEvents revents;
- ErtsPollEvents revent_mask;
-
- revent_mask = ~(ERTS_POLL_EV_IN|ERTS_POLL_EV_OUT);
- revent_mask |= state->events;
- revents = pollres[i].events & revent_mask;
-
- if (revents & ERTS_POLL_EV_ERR) {
- /*
- * Let the driver handle the error condition. Only input,
- * only output, or nothing might have been selected.
- * We *do not* want to call a callback that corresponds
- * to an event not selected. revents might give us a clue
- * on which one to call.
- */
- if ((revents & ERTS_POLL_EV_IN)
- || (!(revents & ERTS_POLL_EV_OUT)
- && state->events & ERTS_POLL_EV_IN)) {
- iready(state->driver.select->inport, state, current_cio_time);
- }
- else if (state->events & ERTS_POLL_EV_OUT) {
- oready(state->driver.select->outport, state, current_cio_time);
- }
- }
- else if (revents & (ERTS_POLL_EV_IN|ERTS_POLL_EV_OUT)) {
+ ErtsPollEvents revents = pollres[i].events;
+
+ if (revents & ERTS_POLL_EV_ERR) {
+ /*
+ * Handle error events by triggering all in/out events
+ * that the driver has selected.
+ * We *do not* want to call a callback that corresponds
+ * to an event not selected.
+ */
+ revents = state->events;
+ }
+ else {
+ revents &= (state->events | ERTS_POLL_EV_NVAL);
+ }
+
+ if (revents & (ERTS_POLL_EV_IN|ERTS_POLL_EV_OUT)) {
if (revents & ERTS_POLL_EV_OUT) {
oready(state->driver.select->outport, state, current_cio_time);
}
@@ -1731,7 +2271,7 @@ ERTS_CIO_EXPORT(erts_check_io)(int do_wait)
was read (true also on the non-smp emulator since
oready() may have been called); therefore, update
revents... */
- revents &= ~(~state->events & ERTS_POLL_EV_IN);
+ revents &= state->events;
if (revents & ERTS_POLL_EV_IN) {
iready(state->driver.select->inport, state, current_cio_time);
}
@@ -1746,6 +2286,69 @@ ERTS_CIO_EXPORT(erts_check_io)(int do_wait)
break;
}
+ case ERTS_EV_TYPE_NIF: { /* Requested via enif_select()... */
+ struct erts_nif_select_event in = {NIL};
+ struct erts_nif_select_event out = {NIL};
+ ErlNifResource* resource;
+ ErtsPollEvents revents = pollres[i].events;
+
+ if (revents & ERTS_POLL_EV_ERR) {
+ /*
+ * Handle error events by triggering all in/out events
+ * that the NIF has selected.
+ * We *do not* want to send a message that corresponds
+ * to an event not selected.
+ */
+ revents = state->events;
+ }
+ else {
+ revents &= (state->events | ERTS_POLL_EV_NVAL);
+ }
+
+ if (revents & (ERTS_POLL_EV_IN|ERTS_POLL_EV_OUT)) {
+ if (revents & ERTS_POLL_EV_OUT) {
+ if (is_not_nil(state->driver.nif->out.pid)) {
+ out = state->driver.nif->out;
+ resource = state->driver.stop.resource;
+ state->driver.nif->out.ddeselect_cnt = ERTS_NIF_DELAYED_DESELECT;
+ state->driver.nif->out.pid = NIL;
+ add_active_fd(state->fd);
+ }
+ else {
+ ASSERT(state->driver.nif->out.ddeselect_cnt >= 2);
+ state->driver.nif->out.ddeselect_cnt--;
+ }
+ }
+ if (revents & ERTS_POLL_EV_IN) {
+ if (is_not_nil(state->driver.nif->in.pid)) {
+ in = state->driver.nif->in;
+ resource = state->driver.stop.resource;
+ state->driver.nif->in.ddeselect_cnt = ERTS_NIF_DELAYED_DESELECT;
+ state->driver.nif->in.pid = NIL;
+ add_active_fd(state->fd);
+ }
+ else {
+ ASSERT(state->driver.nif->in.ddeselect_cnt >= 2);
+ state->driver.nif->in.ddeselect_cnt--;
+ }
+ }
+ }
+ else if (revents & ERTS_POLL_EV_NVAL) {
+ abort();
+ }
+
+#ifdef ERTS_SMP
+ erts_smp_mtx_unlock(fd_mtx(fd));
+#endif
+ if (is_not_nil(in.pid)) {
+ send_event_tuple(&in, resource, am_ready_input);
+ }
+ if (is_not_nil(out.pid)) {
+ send_event_tuple(&out, resource, am_ready_output);
+ }
+ goto next_pollres_unlocked;
+ }
+
#if ERTS_CIO_HAVE_DRV_EVENT
case ERTS_EV_TYPE_DRV_EV: { /* Requested via driver_event()... */
ErlDrvEventData event_data;
@@ -1786,6 +2389,7 @@ ERTS_CIO_EXPORT(erts_check_io)(int do_wait)
#ifdef ERTS_SMP
erts_smp_mtx_unlock(fd_mtx(fd));
#endif
+ next_pollres_unlocked:;
}
erts_smp_atomic_set_nob(&pollset.in_poll_wait, 0);
@@ -2311,6 +2915,39 @@ static void doit_erts_check_io_debug(void *vstate, void *vcounters)
}
}
}
+ else if (state->type == ERTS_EV_TYPE_NIF) {
+ ErlNifResource* r;
+ erts_printf("enif_select ");
+
+#ifdef ERTS_SYS_CONTINOUS_FD_NUMBERS
+ if (internal) {
+ erts_printf("internal ");
+ err = 1;
+ }
+
+ if (cio_events == ep_events) {
+ erts_printf("ev=");
+ if (print_events(cio_events) != 0)
+ err = 1;
+ }
+ else {
+ err = 1;
+ erts_printf("cio_ev=");
+ print_events(cio_events);
+ erts_printf(" ep_ev=");
+ print_events(ep_events);
+ }
+#else
+ if (print_events(cio_events) != 0)
+ err = 1;
+#endif
+ erts_printf(" inpid=%T dd_cnt=%b32d", state->driver.nif->in.pid,
+ state->driver.nif->in.ddeselect_cnt);
+ erts_printf(" outpid=%T dd_cnt=%b32d", state->driver.nif->out.pid,
+ state->driver.nif->out.ddeselect_cnt);
+ r = state->driver.stop.resource;
+ erts_printf(" resource=%p(%T:%T)", r, r->type->module, r->type->name);
+ }
#if ERTS_CIO_HAVE_DRV_EVENT
else if (state->type == ERTS_EV_TYPE_DRV_EV) {
Eterm id;
@@ -2367,11 +3004,12 @@ static void doit_erts_check_io_debug(void *vstate, void *vcounters)
erts_printf("control_type=%d ", (int)state->type);
#ifdef ERTS_SYS_CONTINOUS_FD_NUMBERS
if (cio_events == ep_events) {
- erts_printf("ev=0x%b32x", (Uint32) cio_events);
+ erts_printf("ev=");
+ print_events(cio_events);
}
else {
- erts_printf("cio_ev=0x%b32x", (Uint32) cio_events);
- erts_printf(" ep_ev=0x%b32x", (Uint32) ep_events);
+ erts_printf("cio_ev="); print_events(cio_events);
+ erts_printf(" ep_ev="); print_events(ep_events);
}
#else
erts_printf("ev=0x%b32x", (Uint32) cio_events);
@@ -2400,7 +3038,7 @@ ERTS_CIO_EXPORT(erts_check_io_debug)(ErtsCheckIoDebugInfo *ciodip)
#if ERTS_CIO_HAVE_DRV_EVENT
null_des.driver.event = NULL;
#endif
- null_des.driver.drv_ptr = NULL;
+ null_des.driver.stop.drv_ptr = NULL;
null_des.events = 0;
null_des.remove_cnt = 0;
null_des.type = ERTS_EV_TYPE_NONE;
diff --git a/erts/emulator/sys/common/erl_check_io.h b/erts/emulator/sys/common/erl_check_io.h
index 14f1ea3f43..41e2e9210a 100644
--- a/erts/emulator/sys/common/erl_check_io.h
+++ b/erts/emulator/sys/common/erl_check_io.h
@@ -34,6 +34,8 @@
int driver_select_kp(ErlDrvPort, ErlDrvEvent, int, int);
int driver_select_nkp(ErlDrvPort, ErlDrvEvent, int, int);
+enum ErlNifSelectReturn enif_select_kp(ErlNifEnv*, ErlNifEvent, enum ErlNifSelectFlags, void*, Eterm);
+enum ErlNifSelectReturn enif_select_nkp(ErlNifEnv*, ErlNifEvent, enum ErlNifSelectFlags, void*, Eterm);
int driver_event_kp(ErlDrvPort, ErlDrvEvent, ErlDrvEventData);
int driver_event_nkp(ErlDrvPort, ErlDrvEvent, ErlDrvEventData);
Uint erts_check_io_size_kp(void);
@@ -136,4 +138,20 @@ typedef struct {
ErtsIoTask iniotask;
ErtsIoTask outiotask;
} ErtsDrvSelectDataState;
+
+struct erts_nif_select_event {
+ Eterm pid;
+ Eterm immed;
+ Uint32 refn[ERTS_REF_NUMBERS];
+ Sint32 ddeselect_cnt; /* 0: No delayed deselect in progress
+ * 1: Do deselect before next poll
+ * >1: Countdown of ignored events
+ */
+};
+
+typedef struct {
+ struct erts_nif_select_event in;
+ struct erts_nif_select_event out;
+} ErtsNifSelectDataState;
+
#endif /* #ifndef ERL_CHECK_IO_INTERNAL__ */
diff --git a/erts/emulator/sys/common/erl_poll.c b/erts/emulator/sys/common/erl_poll.c
index b8a28bcc18..0c4df3e13a 100644
--- a/erts/emulator/sys/common/erl_poll.c
+++ b/erts/emulator/sys/common/erl_poll.c
@@ -3030,6 +3030,7 @@ ERTS_POLL_EXPORT(erts_poll_get_selected_events)(ErtsPollSet ps,
ev[fd] = 0;
else {
ev[fd] = ps->fds_status[fd].events;
+ ASSERT(ps->fds_status[fd].used_events == ev[fd]);
if (
#if ERTS_POLL_USE_WAKEUP_PIPE
fd == ps->wake_fds[0] || fd == ps->wake_fds[1] ||
diff --git a/erts/emulator/sys/unix/sys.c b/erts/emulator/sys/unix/sys.c
index 2fc802a2c6..a852550915 100644
--- a/erts/emulator/sys/unix/sys.c
+++ b/erts/emulator/sys/unix/sys.c
@@ -161,6 +161,7 @@ int erts_use_kernel_poll = 0;
struct {
int (*select)(ErlDrvPort, ErlDrvEvent, int, int);
+ enum ErlNifSelectReturn (*enif_select)(ErlNifEnv*, ErlNifEvent, enum ErlNifSelectFlags, void*, Eterm);
int (*event)(ErlDrvPort, ErlDrvEvent, ErlDrvEventData);
void (*check_io_as_interrupt)(void);
void (*check_io_interrupt)(int);
@@ -184,6 +185,13 @@ driver_event(ErlDrvPort port, ErlDrvEvent event, ErlDrvEventData event_data)
return (*io_func.event)(port, event, event_data);
}
+int enif_select(ErlNifEnv* env, ErlNifEvent event,
+ enum ErlNifSelectFlags flags, void* obj, Eterm ref)
+{
+ return (*io_func.enif_select)(env, event, flags, obj, ref);
+}
+
+
Eterm erts_check_io_info(void *p)
{
return (*io_func.info)(p);
@@ -201,6 +209,7 @@ init_check_io(void)
{
if (erts_use_kernel_poll) {
io_func.select = driver_select_kp;
+ io_func.enif_select = enif_select_kp;
io_func.event = driver_event_kp;
#ifdef ERTS_POLL_NEED_ASYNC_INTERRUPT_SUPPORT
io_func.check_io_as_interrupt = erts_check_io_async_sig_interrupt_kp;
@@ -216,6 +225,7 @@ init_check_io(void)
}
else {
io_func.select = driver_select_nkp;
+ io_func.enif_select = enif_select_nkp;
io_func.event = driver_event_nkp;
#ifdef ERTS_POLL_NEED_ASYNC_INTERRUPT_SUPPORT
io_func.check_io_as_interrupt = erts_check_io_async_sig_interrupt_nkp;
diff --git a/erts/emulator/test/nif_SUITE.erl b/erts/emulator/test/nif_SUITE.erl
index 36d512e388..27276e6646 100644
--- a/erts/emulator/test/nif_SUITE.erl
+++ b/erts/emulator/test/nif_SUITE.erl
@@ -31,6 +31,7 @@
init_per_testcase/2, end_per_testcase/2,
basic/1, reload_error/1, upgrade/1, heap_frag/1,
t_on_load/1,
+ select/1,
hipe/1,
types/1, many_args/1, binaries/1, get_string/1, get_atom/1,
maps/1,
@@ -64,6 +65,7 @@ all() ->
[{group, G} || G <- api_groups()]
++
[reload_error, heap_frag, types, many_args,
+ select,
hipe,
binaries, get_string, get_atom, maps, api_macros, from_array,
iolist_as_binary, resource, resource_binary,
@@ -434,6 +436,129 @@ t_on_load(Config) when is_list(Config) ->
verify_tmpmem(TmpMem),
ok.
+-define(ERL_NIF_SELECT_READ, (1 bsl 0)).
+-define(ERL_NIF_SELECT_WRITE, (1 bsl 1)).
+-define(ERL_NIF_SELECT_STOP, (1 bsl 2)).
+
+-define(ERL_NIF_SELECT_ERROR, (1 bsl 0)).
+-define(ERL_NIF_SELECT_STOP_CALLED, (1 bsl 1)).
+-define(ERL_NIF_SELECT_STOP_SCHEDULED, (1 bsl 2)).
+-define(ERL_NIF_SELECT_INVALID_EVENT, (1 bsl 3)).
+-define(ERL_NIF_SELECT_FAILED, (1 bsl 4)).
+
+
+select(Config) when is_list(Config) ->
+ ensure_lib_loaded(Config),
+
+ Ref = make_ref(),
+ {{R, R_ptr}, {W, W_ptr}} = pipe_nif(),
+ ok = write_nif(W, <<"hej">>),
+ <<"hej">> = read_nif(R, 3),
+
+ %% Wait for read
+ eagain = read_nif(R, 3),
+ 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,Ref),
+ [] = flush(),
+ ok = write_nif(W, <<"hej">>),
+ [{select, R, Ref, ready_input}] = flush(),
+ <<"hej">> = read_nif(R, 3),
+
+ %% Wait for write
+ Written = write_full(W, $a),
+ 0 = select_nif(W,?ERL_NIF_SELECT_WRITE,W,Ref),
+ [] = flush(),
+ Half = byte_size(Written) div 2,
+ <<First:Half/binary,Second/binary>> = Written,
+ First = read_nif(R,Half),
+ [{select, W, Ref, ready_output}] = flush(),
+ Third = write_full(W, $A),
+ Half2 = byte_size(Second),
+ <<Second:Half2/binary, Third/binary>> = read_nif(R, byte_size(Written)),
+
+ %% Close write and wait for EOF
+ eagain = read_nif(R, 1),
+ check_stop_ret(select_nif(W,?ERL_NIF_SELECT_STOP,W,Ref)),
+ [{fd_resource_stop, W_ptr, _}] = flush(),
+ {1, {W_ptr,_}} = last_fd_stop_call(),
+ true = is_closed_nif(W),
+ [] = flush(),
+ 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,Ref),
+ [{select, R, Ref, ready_input}] = flush(),
+ eof = read_nif(R,1),
+
+ check_stop_ret(select_nif(R,?ERL_NIF_SELECT_STOP,R,Ref)),
+ [{fd_resource_stop, R_ptr, _}] = flush(),
+ {1, {R_ptr,_}} = last_fd_stop_call(),
+ true = is_closed_nif(R),
+
+ select_2(Config).
+
+select_2(Config) ->
+ erlang:garbage_collect(),
+ {_,_,2} = last_resource_dtor_call(),
+
+ Ref1 = make_ref(),
+ Ref2 = make_ref(),
+ {{R, R_ptr}, {W, W_ptr}} = pipe_nif(),
+
+ %% Change ref
+ eagain = read_nif(R, 1),
+ 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,Ref1),
+ 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,Ref2),
+
+ [] = flush(),
+ ok = write_nif(W, <<"hej">>),
+ [{select, R, Ref2, ready_input}] = flush(),
+ <<"hej">> = read_nif(R, 3),
+
+ %% Change pid
+ eagain = read_nif(R, 1),
+ 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,Ref1),
+ Papa = self(),
+ Pid2 = spawn_link(fun() ->
+ 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,Ref1),
+ [] = flush(),
+ Papa ! sync,
+ [{select, R, Ref1, ready_input}] = flush(),
+ <<"hej">> = read_nif(R, 3),
+ Papa ! done
+ end),
+ sync = receive_any(),
+ ok = write_nif(W, <<"hej">>),
+ done = receive_any(),
+ [] = flush(),
+
+ check_stop_ret(select_nif(R,?ERL_NIF_SELECT_STOP,R,Ref1)),
+ [{fd_resource_stop, R_ptr, _}] = flush(),
+ {1, {R_ptr,_}} = last_fd_stop_call(),
+ true = is_closed_nif(R),
+
+ %% Stop without previous read/write select
+ ?ERL_NIF_SELECT_STOP_CALLED = select_nif(W,?ERL_NIF_SELECT_STOP,W,Ref1),
+ [{fd_resource_stop, W_ptr, 1}] = flush(),
+ {1, {W_ptr,1}} = last_fd_stop_call(),
+ true = is_closed_nif(W),
+
+ select_3(Config).
+
+select_3(Config) ->
+ erlang:garbage_collect(),
+ {_,_,2} = last_resource_dtor_call(),
+ ok.
+
+check_stop_ret(?ERL_NIF_SELECT_STOP_CALLED) -> ok;
+check_stop_ret(?ERL_NIF_SELECT_STOP_SCHEDULED) -> ok.
+
+write_full(W, C) ->
+ write_full(W, C, <<>>).
+write_full(W, C, Acc) ->
+ case write_nif(W, <<C>>) of
+ ok ->
+ write_full(W, (C+1) band 255, <<Acc/binary, C>>);
+ {eagain,0} ->
+ Acc
+ end.
+
hipe(Config) when is_list(Config) ->
Data = proplists:get_value(data_dir, Config),
Priv = proplists:get_value(priv_dir, Config),
@@ -1910,6 +2035,15 @@ call(Pid,Cmd) ->
receive_any() ->
receive M -> M end.
+flush() ->
+ flush(10).
+flush(Timeout) ->
+ receive M ->
+ [M | flush(Timeout)]
+ after Timeout ->
+ []
+ end.
+
repeat(0, _, Arg) ->
Arg;
repeat(N, Fun, Arg0) ->
@@ -2273,6 +2407,12 @@ term_to_binary_nif(_, _) -> ?nif_stub.
binary_to_term_nif(_, _, _) -> ?nif_stub.
port_command_nif(_, _) -> ?nif_stub.
format_term_nif(_,_) -> ?nif_stub.
+select_nif(_,_,_,_) -> ?nif_stub.
+pipe_nif() -> ?nif_stub.
+write_nif(_,_) -> ?nif_stub.
+read_nif(_,_) -> ?nif_stub.
+is_closed_nif(_) -> ?nif_stub.
+last_fd_stop_call() -> ?nif_stub.
%% maps
is_map_nif(_) -> ?nif_stub.
diff --git a/erts/emulator/test/nif_SUITE_data/nif_SUITE.c b/erts/emulator/test/nif_SUITE_data/nif_SUITE.c
index 2c93891852..c4f9611ec8 100644
--- a/erts/emulator/test/nif_SUITE_data/nif_SUITE.c
+++ b/erts/emulator/test/nif_SUITE_data/nif_SUITE.c
@@ -25,6 +25,8 @@
#include <limits.h>
#ifndef __WIN32__
#include <unistd.h>
+#include <fcntl.h>
+#include <errno.h>
#endif
#include "nif_mod.h"
@@ -42,7 +44,10 @@ static ERL_NIF_TERM atom_second;
static ERL_NIF_TERM atom_millisecond;
static ERL_NIF_TERM atom_microsecond;
static ERL_NIF_TERM atom_nanosecond;
-
+static ERL_NIF_TERM atom_eagain;
+static ERL_NIF_TERM atom_eof;
+static ERL_NIF_TERM atom_error;
+static ERL_NIF_TERM atom_fd_resource_stop;
typedef struct
{
@@ -102,6 +107,20 @@ struct binary_resource {
unsigned size;
};
+static ErlNifResourceType* fd_resource_type;
+static void fd_resource_dtor(ErlNifEnv* env, void* obj);
+static void fd_resource_stop(ErlNifEnv* env, void* obj, ErlNifEvent, int);
+static ErlNifResourceTypeInit fd_rt_init = {
+ fd_resource_dtor,
+ fd_resource_stop
+};
+struct fd_resource {
+ int fd;
+ int was_selected;
+ ErlNifPid pid;
+};
+
+
static int get_pointer(ErlNifEnv* env, ERL_NIF_TERM term, void** pp)
{
ErlNifBinary bin;
@@ -116,7 +135,6 @@ static ERL_NIF_TERM make_pointer(ErlNifEnv* env, void* p)
{
void** bin_data;
ERL_NIF_TERM res;
- ADD_CALL("get_priv_data_ptr");
bin_data = (void**)enif_make_new_binary(env, sizeof(void*), &res);
*bin_data = p;
return res;
@@ -144,6 +162,9 @@ static int load(ErlNifEnv* env, void** priv_data, ERL_NIF_TERM load_info)
msgenv_resource_type = enif_open_resource_type(env,NULL,"nif_SUITE.msgenv",
msgenv_dtor,
ERL_NIF_RT_CREATE, NULL);
+ fd_resource_type = enif_open_resource_type_x(env, "nif_SUITE.fd",
+ &fd_rt_init,
+ ERL_NIF_RT_CREATE, NULL);
atom_false = enif_make_atom(env,"false");
atom_true = enif_make_atom(env,"true");
atom_self = enif_make_atom(env,"self");
@@ -154,6 +175,10 @@ static int load(ErlNifEnv* env, void** priv_data, ERL_NIF_TERM load_info)
atom_millisecond = enif_make_atom(env,"millisecond");
atom_microsecond = enif_make_atom(env,"microsecond");
atom_nanosecond = enif_make_atom(env,"nanosecond");
+ atom_eagain = enif_make_atom(env, "eagain");
+ atom_eof = enif_make_atom(env, "eof");
+ atom_error = enif_make_atom(env, "error");
+ atom_fd_resource_stop = enif_make_atom(env, "fd_resource_stop");
*priv_data = data;
return 0;
@@ -389,8 +414,7 @@ static ERL_NIF_TERM type_test(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[
ErlNifSInt64 sint64;
ErlNifUInt64 uint64;
double d;
- ERL_NIF_TERM atom, ref1, ref2, term;
- size_t len;
+ ERL_NIF_TERM atom, ref1, ref2;
sint = INT_MIN;
do {
@@ -1024,6 +1048,7 @@ struct make_term_info
{
ErlNifEnv* caller_env;
ErlNifEnv* dst_env;
+ int dst_env_valid;
ERL_NIF_TERM reuse[MAKE_TERM_REUSE_LEN];
unsigned reuse_push;
unsigned reuse_pull;
@@ -1053,6 +1078,7 @@ static ERL_NIF_TERM pull_term(struct make_term_info* mti)
mti->reuse_push < MAKE_TERM_REUSE_LEN) {
mti->reuse_pull = 0;
if (mti->reuse_push == 0) {
+ assert(mti->dst_env_valid);
mti->reuse[0] = enif_make_list(mti->dst_env, 0);
}
}
@@ -1106,10 +1132,6 @@ static ERL_NIF_TERM make_term_string(struct make_term_info* mti, int n)
{
return enif_make_string(mti->dst_env, "Hello!", ERL_NIF_LATIN1);
}
-static ERL_NIF_TERM make_term_ref(struct make_term_info* mti, int n)
-{
- return enif_make_ref(mti->dst_env);
-}
static ERL_NIF_TERM make_term_sub_binary(struct make_term_info* mti, int n)
{
ERL_NIF_TERM orig;
@@ -1217,7 +1239,6 @@ static Make_term_Func* make_funcs[] = {
make_term_atom,
make_term_existing_atom,
make_term_string,
- //make_term_ref,
make_term_sub_binary,
make_term_uint,
make_term_long,
@@ -1241,6 +1262,7 @@ static unsigned num_of_make_funcs()
static int make_term_n(struct make_term_info* mti, int n, ERL_NIF_TERM* res)
{
if (n < num_of_make_funcs()) {
+ assert(mti->dst_env_valid);
*res = make_funcs[n](mti, n);
push_term(mti, *res);
return 1;
@@ -1257,6 +1279,7 @@ static ERL_NIF_TERM make_blob(ErlNifEnv* caller_env, ErlNifEnv* dst_env,
struct make_term_info mti;
mti.caller_env = caller_env;
mti.dst_env = dst_env;
+ mti.dst_env_valid = 1;
mti.reuse_push = 0;
mti.reuse_pull = 0;
mti.resource_type = priv->rt_arr[0].t;
@@ -1297,6 +1320,7 @@ static ERL_NIF_TERM alloc_msgenv(ErlNifEnv* env, int argc, const ERL_NIF_TERM ar
sizeof(*mti));
mti->caller_env = NULL;
mti->dst_env = enif_alloc_env();
+ mti->dst_env_valid = 1;
mti->reuse_push = 0;
mti->reuse_pull = 0;
mti->resource_type = priv->rt_arr[0].t;
@@ -1328,6 +1352,7 @@ static ERL_NIF_TERM clear_msgenv(ErlNifEnv* env, int argc, const ERL_NIF_TERM ar
return enif_make_badarg(env);
}
enif_clear_env(mti.p->dst_env);
+ mti.p->dst_env_valid = 1;
mti.p->reuse_pull = 0;
mti.p->reuse_push = 0;
mti.p->blob = enif_make_list(mti.p->dst_env, 0);
@@ -1362,6 +1387,8 @@ static ERL_NIF_TERM send_blob(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[
}
copy = enif_make_copy(env, mti.p->blob);
res = enif_send(env, &to, mti.p->dst_env, mti.p->blob);
+ if (res)
+ mti.p->dst_env_valid = 0;
return enif_make_tuple3(env, atom_ok, enif_make_int(env,res), copy);
}
@@ -1369,7 +1396,6 @@ static ERL_NIF_TERM send3_blob(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv
{
mti_t mti;
ErlNifPid to;
- ERL_NIF_TERM copy;
int res;
if (!enif_get_resource(env, argv[0], msgenv_resource_type, &mti.vp)
|| !enif_get_local_pid(env, argv[1], &to)) {
@@ -1379,6 +1405,8 @@ static ERL_NIF_TERM send3_blob(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv
enif_make_copy(mti.p->dst_env, argv[2]),
mti.p->blob);
res = enif_send(env, &to, mti.p->dst_env, mti.p->blob);
+ if (res)
+ mti.p->dst_env_valid = 0;
return enif_make_int(env,res);
}
@@ -1395,6 +1423,8 @@ void* threaded_sender(void *arg)
mti.p->send_it = 0;
enif_mutex_unlock(mti.p->mtx);
mti.p->send_res = enif_send(NULL, &mti.p->to_pid, mti.p->dst_env, mti.p->blob);
+ if (mti.p->send_res)
+ mti.p->dst_env_valid = 0;
return NULL;
}
@@ -2010,6 +2040,202 @@ static ERL_NIF_TERM format_term(ErlNifEnv* env, int argc, const ERL_NIF_TERM arg
}
+static int get_fd(ErlNifEnv* env, ERL_NIF_TERM term, struct fd_resource** rsrc)
+{
+ if (!enif_get_resource(env, term, fd_resource_type, (void**)rsrc)) {
+ return 0;
+ }
+ return 1;
+}
+
+static ERL_NIF_TERM select_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+ struct fd_resource* fdr;
+ enum ErlNifSelectFlags mode;
+ void* obj;
+ ERL_NIF_TERM ref;
+ enum ErlNifSelectReturn retval;
+
+ if (!get_fd(env, argv[0], &fdr)
+ || !enif_get_uint(env, argv[1], &mode)
+ || !enif_get_resource(env, argv[2], fd_resource_type, &obj))
+ {
+ return enif_make_badarg(env);
+ }
+
+ ref = argv[3];
+
+ fdr->was_selected = 1;
+ enif_self(env, &fdr->pid);
+ retval = enif_select(env, fdr->fd, mode, obj, ref);
+
+ return enif_make_int(env, (int)retval);
+}
+
+static ERL_NIF_TERM pipe_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+ struct fd_resource* read_rsrc;
+ struct fd_resource* write_rsrc;
+ ERL_NIF_TERM read_fd, write_fd;
+ int fds[2], flags;
+
+ if (pipe(fds) < 0)
+ return enif_make_string(env, "pipe failed", ERL_NIF_LATIN1);
+
+ if ((flags = fcntl(fds[0], F_GETFL, 0)) < 0
+ || fcntl(fds[0], F_SETFL, flags|O_NONBLOCK) < 0
+ || (flags = fcntl(fds[1], F_GETFL, 0)) < 0
+ || fcntl(fds[1], F_SETFL, flags|O_NONBLOCK) < 0) {
+ close(fds[0]);
+ close(fds[1]);
+ return enif_make_string(env, "fcntl failed on pipe", ERL_NIF_LATIN1);
+ }
+
+ read_rsrc = enif_alloc_resource(fd_resource_type, sizeof(struct fd_resource));
+ write_rsrc = enif_alloc_resource(fd_resource_type, sizeof(struct fd_resource));
+ read_rsrc->fd = fds[0];
+ write_rsrc->fd = fds[1];
+ read_fd = enif_make_resource(env, read_rsrc);
+ write_fd = enif_make_resource(env, write_rsrc);
+ enif_release_resource(read_rsrc);
+ enif_release_resource(write_rsrc);
+
+ return enif_make_tuple2(env,
+ enif_make_tuple2(env, read_fd, make_pointer(env, read_rsrc)),
+ enif_make_tuple2(env, write_fd, make_pointer(env, write_rsrc)));
+}
+
+static ERL_NIF_TERM write_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+ struct fd_resource* fdr;
+ ErlNifBinary bin;
+ int n, written = 0;
+
+ if (!get_fd(env, argv[0], &fdr)
+ || !enif_inspect_binary(env, argv[1], &bin))
+ return enif_make_badarg(env);
+
+ for (;;) {
+ n = write(fdr->fd, bin.data + written, bin.size - written);
+ if (n >= 0) {
+ written += n;
+ if (written == bin.size) {
+ return atom_ok;
+ }
+ }
+ else if (errno == EAGAIN) {
+ return enif_make_tuple2(env, atom_eagain, enif_make_int(env, written));
+ }
+ else if (errno == EINTR) {
+ continue;
+ }
+ else {
+ return enif_make_tuple2(env, atom_error, enif_make_int(env, errno));
+ }
+ }
+}
+
+static ERL_NIF_TERM read_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+ struct fd_resource* fdr;
+ unsigned char* buf;
+ int n, count;
+ ERL_NIF_TERM res;
+
+ if (!get_fd(env, argv[0], &fdr)
+ || !enif_get_int(env, argv[1], &count) || count < 1)
+ return enif_make_badarg(env);
+
+ buf = enif_make_new_binary(env, count, &res);
+
+ for (;;) {
+ n = read(fdr->fd, buf, count);
+ if (n > 0) {
+ if (n < count) {
+ res = enif_make_sub_binary(env, res, 0, n);
+ }
+ return res;
+ }
+ else if (n == 0) {
+ return atom_eof;
+ }
+ else if (errno == EAGAIN) {
+ return atom_eagain;
+ }
+ else if (errno == EINTR) {
+ continue;
+ }
+ else {
+ return enif_make_tuple2(env, atom_error, enif_make_int(env, errno));
+ }
+ }
+}
+
+static ERL_NIF_TERM is_closed_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+ struct fd_resource* fdr;
+
+ if (!get_fd(env, argv[0], &fdr))
+ return enif_make_badarg(env);
+
+ return fdr->fd < 0 ? atom_true : atom_false;
+}
+
+static void fd_resource_dtor(ErlNifEnv* env, void* obj)
+{
+ struct fd_resource* fdr = (struct fd_resource*)obj;
+ resource_dtor(env, obj);
+ if (fdr->fd >= 0) {
+ assert(!fdr->was_selected);
+ close(fdr->fd);
+ }
+}
+
+static struct {
+ void* obj;
+ int was_direct_call;
+}last_fd_stop;
+int fd_stop_cnt = 0;
+
+static void fd_resource_stop(ErlNifEnv* env, void* obj, ErlNifEvent fd,
+ int is_direct_call)
+{
+ struct fd_resource* fdr = (struct fd_resource*)obj;
+ assert(fd == fdr->fd);
+ assert(fd >= 0);
+
+ last_fd_stop.obj = obj;
+ last_fd_stop.was_direct_call = is_direct_call;
+ fd_stop_cnt++;
+
+ close(fd);
+ fdr->fd = -1; /* thread safety ? */
+ fdr->was_selected = 0;
+
+ {
+ ErlNifEnv* msg_env = enif_alloc_env();
+ ERL_NIF_TERM msg;
+ msg = enif_make_tuple3(msg_env,
+ atom_fd_resource_stop,
+ make_pointer(msg_env, obj),
+ enif_make_int(msg_env, is_direct_call));
+
+ enif_send(env, &fdr->pid, msg_env, msg);
+ enif_free_env(msg_env);
+ }
+}
+
+static ERL_NIF_TERM last_fd_stop_call(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+ ERL_NIF_TERM last, ret;
+ last = enif_make_tuple2(env, make_pointer(env, last_fd_stop.obj),
+ enif_make_int(env, last_fd_stop.was_direct_call));
+ ret = enif_make_tuple2(env, enif_make_int(env, fd_stop_cnt), last);
+ fd_stop_cnt = 0;
+ return ret;
+}
+
+
static ErlNifFunc nif_funcs[] =
{
{"lib_version", 0, lib_version},
@@ -2086,7 +2312,13 @@ static ErlNifFunc nif_funcs[] =
{"term_to_binary_nif", 2, term_to_binary},
{"binary_to_term_nif", 3, binary_to_term},
{"port_command_nif", 2, port_command},
- {"format_term_nif", 2, format_term}
+ {"format_term_nif", 2, format_term},
+ {"select_nif", 4, select_nif},
+ {"pipe_nif", 0, pipe_nif},
+ {"write_nif", 2, write_nif},
+ {"read_nif", 2, read_nif},
+ {"is_closed_nif", 1, is_closed_nif},
+ {"last_fd_stop_call", 0, last_fd_stop_call}
};
ERL_NIF_INIT(nif_SUITE,nif_funcs,load,NULL,upgrade,unload)
generated by cgit v1.2.3 (git 2.25.1) at 2025-01-03 06:22:27 +0000