aboutsummaryrefslogtreecommitdiffstats
path: root/erts
diff options
context:
space:
mode:
Diffstat (limited to 'erts')
-rw-r--r--erts/emulator/nifs/common/socket_nif.c798
-rw-r--r--erts/preloaded/ebin/socket.beambin66152 -> 66040 bytes
-rw-r--r--erts/preloaded/src/socket.erl126
3 files changed, 800 insertions, 124 deletions
diff --git a/erts/emulator/nifs/common/socket_nif.c b/erts/emulator/nifs/common/socket_nif.c
index 6564c3c82f..c48d6eab00 100644
--- a/erts/emulator/nifs/common/socket_nif.c
+++ b/erts/emulator/nifs/common/socket_nif.c
@@ -104,6 +104,14 @@
#undef WANT_NONBLOCKING
#include "sys.h"
+
+
+
+/* AND HERE WE MAY HAVE A BUNCH OF DEFINES....SEE INET DRIVER.... */
+
+
+
+
#else /* !__WIN32__ */
#include <sys/time.h>
@@ -297,7 +305,7 @@ static void (*esock_sctp_freepaddrs)(struct sockaddr *addrs) = NULL;
#endif
#include "sys.h"
-#endif
+#endif /* !__WIN32__ */
#include <erl_nif.h>
@@ -731,6 +739,7 @@ typedef struct {
SocketAddress remote;
unsigned int addrLen;
+ ErlNifEnv* env;
/* +++ Controller (owner) process +++ */
ErlNifPid ctrlPid;
@@ -771,7 +780,7 @@ typedef struct {
size_t rBufSz; // Read buffer size (when data length = 0 is specified)
size_t rCtrlSz; // Read control buffer size
size_t wCtrlSz; // Write control buffer size
- BOOLEAN_T iow; // Inform On Wrap
+ BOOLEAN_T iow; // Inform On (counter) Wrap
BOOLEAN_T dbg;
/* +++ Close stuff +++ */
@@ -909,11 +918,9 @@ static ERL_NIF_TERM nif_finalize_connection(ErlNifEnv* env,
static ERL_NIF_TERM nif_finalize_close(ErlNifEnv* env,
int argc,
const ERL_NIF_TERM argv[]);
-/*
static ERL_NIF_TERM nif_cancel(ErlNifEnv* env,
int argc,
const ERL_NIF_TERM argv[]);
-*/
static ERL_NIF_TERM nopen(ErlNifEnv* env,
@@ -977,7 +984,6 @@ static ERL_NIF_TERM nclose(ErlNifEnv* env,
static ERL_NIF_TERM nshutdown(ErlNifEnv* env,
SocketDescriptor* descP,
int how);
-
static ERL_NIF_TERM nsetopt(ErlNifEnv* env,
SocketDescriptor* descP,
BOOLEAN_T isEncoded,
@@ -1817,6 +1823,43 @@ static ERL_NIF_TERM nsockname(ErlNifEnv* env,
SocketDescriptor* descP);
static ERL_NIF_TERM npeername(ErlNifEnv* env,
SocketDescriptor* descP);
+static ERL_NIF_TERM ncancel(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM op,
+ ERL_NIF_TERM opRef);
+static ERL_NIF_TERM ncancel_connect(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM opRef);
+static ERL_NIF_TERM ncancel_accept(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM opRef);
+static ERL_NIF_TERM ncancel_accept_current(ErlNifEnv* env,
+ SocketDescriptor* descP);
+static ERL_NIF_TERM ncancel_accept_waiting(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM opRef);
+static ERL_NIF_TERM ncancel_send(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM opRef);
+static ERL_NIF_TERM ncancel_send_current(ErlNifEnv* env,
+ SocketDescriptor* descP);
+static ERL_NIF_TERM ncancel_send_waiting(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM opRef);
+static ERL_NIF_TERM ncancel_recv(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM opRef);
+static ERL_NIF_TERM ncancel_read_select(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM opRef);
+static ERL_NIF_TERM ncancel_write_select(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM opRef);
+static ERL_NIF_TERM ncancel_mode_select(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM opRef,
+ int smode,
+ int rmode);
static ERL_NIF_TERM nsetopt_str_opt(ErlNifEnv* env,
SocketDescriptor* descP,
@@ -1858,6 +1901,10 @@ static ERL_NIF_TERM ngetopt_timeval_opt(ErlNifEnv* env,
int level,
int opt);
+static BOOLEAN_T send_check_writer(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM ref,
+ ERL_NIF_TERM* checkResult);
static ERL_NIF_TERM send_check_result(ErlNifEnv* env,
SocketDescriptor* descP,
ssize_t written,
@@ -2083,6 +2130,25 @@ static BOOLEAN_T acceptor_pop(ErlNifEnv* env,
ErlNifPid* pid,
ErlNifMonitor* mon,
ERL_NIF_TERM* ref);
+static BOOLEAN_T acceptor_unqueue(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ const ErlNifPid* pid);
+
+static BOOLEAN_T writer_search4pid(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ErlNifPid* pid);
+static ERL_NIF_TERM writer_push(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ErlNifPid pid,
+ ERL_NIF_TERM ref);
+static BOOLEAN_T writer_pop(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ErlNifPid* pid,
+ ErlNifMonitor* mon,
+ ERL_NIF_TERM* ref);
+static BOOLEAN_T writer_unqueue(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ const ErlNifPid* pid);
static BOOLEAN_T qsearch4pid(ErlNifEnv* env,
SocketRequestQueue* q,
@@ -2235,8 +2301,10 @@ static char str_exsend[] = "exsend"; // failed send
/* *** "Global" Atoms *** */
+ERL_NIF_TERM esock_atom_accept;
ERL_NIF_TERM esock_atom_addr;
ERL_NIF_TERM esock_atom_any;
+ERL_NIF_TERM esock_atom_connect;
ERL_NIF_TERM esock_atom_credentials;
ERL_NIF_TERM esock_atom_ctrl;
ERL_NIF_TERM esock_atom_ctrunc;
@@ -2261,6 +2329,7 @@ ERL_NIF_TERM esock_atom_local;
ERL_NIF_TERM esock_atom_loopback;
ERL_NIF_TERM esock_atom_lowdelay;
ERL_NIF_TERM esock_atom_mincost;
+ERL_NIF_TERM esock_atom_not_found;
ERL_NIF_TERM esock_atom_ok;
ERL_NIF_TERM esock_atom_oob;
ERL_NIF_TERM esock_atom_origdstaddr;
@@ -2270,11 +2339,18 @@ ERL_NIF_TERM esock_atom_port;
ERL_NIF_TERM esock_atom_protocol;
ERL_NIF_TERM esock_atom_raw;
ERL_NIF_TERM esock_atom_rdm;
-ERL_NIF_TERM esock_atom_rights;
+ERL_NIF_TERM esock_atom_recv;
+ERL_NIF_TERM esock_atom_recvfrom;
+ERL_NIF_TERM esock_atom_recvmsg;
ERL_NIF_TERM esock_atom_reliability;
+ERL_NIF_TERM esock_atom_rights;
ERL_NIF_TERM esock_atom_scope_id;
ERL_NIF_TERM esock_atom_sctp;
ERL_NIF_TERM esock_atom_sec;
+ERL_NIF_TERM esock_atom_select_sent;
+ERL_NIF_TERM esock_atom_send;
+ERL_NIF_TERM esock_atom_sendmsg;
+ERL_NIF_TERM esock_atom_sendto;
ERL_NIF_TERM esock_atom_seqpacket;
ERL_NIF_TERM esock_atom_socket;
ERL_NIF_TERM esock_atom_spec_dst;
@@ -3222,7 +3298,8 @@ ERL_NIF_TERM naccept_listening(ErlNifEnv* env,
&descP->currentAcceptor.mon) > 0)
return esock_make_error(env, atom_exmon);
- descP->currentAcceptor.ref = ref;
+ descP->currentAcceptor.ref = enif_make_copy(descP->env, ref);
+ descP->currentAcceptorP = &descP->currentAcceptor;
SELECT(env,
descP->sock,
@@ -3320,7 +3397,8 @@ ERL_NIF_TERM naccept_listening(ErlNifEnv* env,
/* *** naccept_accepting ***
* We have an active acceptor and possibly acceptors waiting in queue.
- * At the moment the queue is *not* implemented.
+ * If the pid of the calling process is not the pid of the "current process",
+ * push the requester onto the queue.
*/
static
ERL_NIF_TERM naccept_accepting(ErlNifEnv* env,
@@ -3347,11 +3425,8 @@ ERL_NIF_TERM naccept_accepting(ErlNifEnv* env,
"\r\n", caller, descP->currentAcceptor.pid) );
if (!compare_pids(env, &descP->currentAcceptor.pid, &caller)) {
- /* This will have to do until we implement the queue.
- * When we have the queue, we should simply push this request,
- * and instead return with eagain (the caller will then wait
- * for the select message).
- */
+
+ /* Not the "current acceptor", so (maybe) push onto queue */
SSDBG( descP, ("SOCKET", "naccept_accepting -> not (active) acceptor\r\n") );
@@ -3411,6 +3486,8 @@ ERL_NIF_TERM naccept_accepting(ErlNifEnv* env,
SSDBG( descP, ("SOCKET", "naccept_accepting -> accept success\r\n") );
+ DEMONP(env, descP, &descP->currentAcceptor.mon);
+
if ((accEvent = sock_create_event(accSock)) == INVALID_EVENT) {
save_errno = sock_errno();
while ((sock_close(accSock) == INVALID_SOCKET) &&
@@ -3451,10 +3528,7 @@ ERL_NIF_TERM naccept_accepting(ErlNifEnv* env,
accDescP->state = SOCKET_STATE_CONNECTED;
- /* Here we should have the test if we have something in the queue.
- * And if so, pop it and copy the (waiting) acceptor, and then
- * make a new select with that info).
- */
+ /* Check if there are waiting acceptors (popping the acceptor queue) */
if (acceptor_pop(env, descP,
&descP->currentAcceptor.pid,
@@ -3559,12 +3633,12 @@ ERL_NIF_TERM nif_send(ErlNifEnv* env,
}
-/* What do we do when another process tries to write
- * when the current writer has a select already waiting?
- * Queue it? And what about simultaneous read and write?
- * Queue up all operations towards the socket?
+/* *** nsend ***
*
- * We (may) need a currentOp field and an ops queue field.
+ * Do the actual send.
+ * Do some initial writer checks, do the actual send and then
+ * analyze the result. If we are done, another writer may be
+ * scheduled (if there is one in the writer queue).
*/
static
ERL_NIF_TERM nsend(ErlNifEnv* env,
@@ -3573,12 +3647,17 @@ ERL_NIF_TERM nsend(ErlNifEnv* env,
ErlNifBinary* sndDataP,
int flags)
{
- int save_errno;
- ssize_t written;
+ int save_errno;
+ ssize_t written;
+ ERL_NIF_TERM writerCheck;
if (!descP->isWritable)
return enif_make_badarg(env);
+ /* Check if there is already a current writer and if its us */
+ if (!send_check_writer(env, descP, sendRef, &writerCheck))
+ return writerCheck;
+
/* We ignore the wrap for the moment.
* Maybe we should issue a wrap-message to controlling process...
*/
@@ -3597,6 +3676,7 @@ ERL_NIF_TERM nsend(ErlNifEnv* env,
}
+
/* ----------------------------------------------------------------------
* nif_sendto
*
@@ -3662,9 +3742,13 @@ ERL_NIF_TERM nif_sendto(ErlNifEnv* env,
&remoteAddrLen)) != NULL)
return esock_make_error_str(env, xres);
+ MLOCK(descP->writeMtx);
+
res = nsendto(env, descP, sendRef, &sndData, flags,
&remoteAddr, remoteAddrLen);
+ MUNLOCK(descP->writeMtx);
+
SGDBG( ("SOCKET", "nif_sendto -> done with result: "
"\r\n %T"
"\r\n", res) );
@@ -3682,12 +3766,17 @@ ERL_NIF_TERM nsendto(ErlNifEnv* env,
SocketAddress* toAddrP,
unsigned int toAddrLen)
{
- int save_errno;
- ssize_t written;
+ int save_errno;
+ ssize_t written;
+ ERL_NIF_TERM writerCheck;
if (!descP->isWritable)
return enif_make_badarg(env);
+ /* Check if there is already a current writer and if its us */
+ if (!send_check_writer(env, descP, sendRef, &writerCheck))
+ return writerCheck;
+
/* We ignore the wrap for the moment.
* Maybe we should issue a wrap-message to controlling process...
*/
@@ -3763,8 +3852,12 @@ ERL_NIF_TERM nif_sendmsg(ErlNifEnv* env,
if (!esendflags2sendflags(eflags, &flags))
return esock_make_error(env, esock_atom_einval);
+ MLOCK(descP->writeMtx);
+
res = nsendmsg(env, descP, sendRef, eMsgHdr, flags);
+ MUNLOCK(descP->writeMtx);
+
SSDBG( descP,
("SOCKET", "nif_sendmsg -> done with result: "
"\r\n %T"
@@ -3791,12 +3884,16 @@ ERL_NIF_TERM nsendmsg(ErlNifEnv* env,
size_t ctrlBufLen, ctrlBufUsed;
int save_errno;
ssize_t written, dataSize;
+ ERL_NIF_TERM writerCheck;
char* xres;
if (!descP->isWritable)
return enif_make_badarg(env);
-
+ /* Check if there is already a current writer and if its us */
+ if (!send_check_writer(env, descP, sendRef, &writerCheck))
+ return writerCheck;
+
/* Depending on if we are *connected* or not, we require
* different things in the msghdr map.
*/
@@ -7175,6 +7272,7 @@ ERL_NIF_TERM nsetopt_lvl_sctp_associnfo(ErlNifEnv* env,
int res;
size_t sz;
unsigned int tmp;
+ int32_t tmpAssocId;
SSDBG( descP,
("SOCKET", "nsetopt_lvl_sctp_associnfo -> entry with"
@@ -7213,8 +7311,12 @@ ERL_NIF_TERM nsetopt_lvl_sctp_associnfo(ErlNifEnv* env,
SSDBG( descP,
("SOCKET", "nsetopt_lvl_sctp_associnfo -> decode attributes\r\n") );
- if (!GET_UINT(env, eAssocId, &assocParams.sasoc_assoc_id))
+ /* On some platforms the assoc id is typed as an unsigned integer (uint32)
+ * So, to avoid warnings there, we always make an explicit cast...
+ */
+ if (!GET_INT(env, eAssocId, &tmpAssocId))
return esock_make_error(env, esock_atom_einval);
+ assocParams.sasoc_assoc_id = (typeof(assocParams.sasoc_assoc_id)) tmpAssocId;
/*
* We should really make sure this is ok in erlang (to ensure that
@@ -7509,6 +7611,7 @@ ERL_NIF_TERM nsetopt_lvl_sctp_rtoinfo(ErlNifEnv* env,
struct sctp_rtoinfo rtoInfo;
int res;
size_t sz;
+ int32_t tmpAssocId;
SSDBG( descP,
("SOCKET", "nsetopt_lvl_sctp_rtoinfo -> entry with"
@@ -7541,8 +7644,12 @@ ERL_NIF_TERM nsetopt_lvl_sctp_rtoinfo(ErlNifEnv* env,
SSDBG( descP,
("SOCKET", "nsetopt_lvl_sctp_rtoinfo -> decode attributes\r\n") );
- if (!GET_UINT(env, eAssocId, &rtoInfo.srto_assoc_id))
+ /* On some platforms the assoc id is typed as an unsigned integer (uint32)
+ * So, to avoid warnings there, we always make an explicit cast...
+ */
+ if (!GET_INT(env, eAssocId, &tmpAssocId))
return esock_make_error(env, esock_atom_einval);
+ rtoInfo.srto_assoc_id = (typeof(rtoInfo.srto_assoc_id)) tmpAssocId;
if (!GET_UINT(env, eInitial, &rtoInfo.srto_initial))
return esock_make_error(env, esock_atom_einval);
@@ -10414,10 +10521,471 @@ ERL_NIF_TERM npeername(ErlNifEnv* env,
/* ----------------------------------------------------------------------
+ * nif_cancel
+ *
+ * Description:
+ * Cancel a previous select!
+ *
+ * Arguments:
+ * Socket (ref) - Points to the socket descriptor.
+ * Operation (atom) - What kind of operation (accept, send, ...) is to be cancelled
+ * Ref (ref) - Unique id for the operation
+ */
+static
+ERL_NIF_TERM nif_cancel(ErlNifEnv* env,
+ int argc,
+ const ERL_NIF_TERM argv[])
+{
+ SocketDescriptor* descP;
+ ERL_NIF_TERM op, opRef, result;
+
+ SGDBG( ("SOCKET", "nif_cancel -> entry with argc: %d\r\n", argc) );
+
+ /* Extract arguments and perform preliminary validation */
+
+ if ((argc != 3) ||
+ !enif_get_resource(env, argv[0], sockets, (void**) &descP)) {
+ return enif_make_badarg(env);
+ }
+ op = argv[1];
+ opRef = argv[2];
+
+ SSDBG( descP,
+ ("SOCKET", "nif_cancel -> args when sock = %d:"
+ "\r\n op: %T"
+ "\r\n opRef: %T"
+ "\r\n", descP->sock, op, opRef) );
+
+ result = ncancel(env, descP, op, opRef);
+
+ SSDBG( descP,
+ ("SOCKET", "nif_cancel -> done with result: "
+ "\r\n %T"
+ "\r\n", result) );
+
+ return result;
+
+}
+
+
+static
+ERL_NIF_TERM ncancel(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM op,
+ ERL_NIF_TERM opRef)
+{
+ /* <KOLLA>
+ *
+ * Do we really need all these variants? Should it not be enough with:
+ *
+ * connect | accept | send | recv
+ *
+ * </KOLLA>
+ */
+ if (COMPARE(op, esock_atom_connect) == 0) {
+ return ncancel_connect(env, descP, opRef);
+ } else if (COMPARE(op, esock_atom_accept) == 0) {
+ return ncancel_accept(env, descP, opRef);
+ } else if (COMPARE(op, esock_atom_send) == 0) {
+ return ncancel_send(env, descP, opRef);
+ } else if (COMPARE(op, esock_atom_sendto) == 0) {
+ return ncancel_send(env, descP, opRef);
+ } else if (COMPARE(op, esock_atom_sendmsg) == 0) {
+ return ncancel_send(env, descP, opRef);
+ } else if (COMPARE(op, esock_atom_recv) == 0) {
+ return ncancel_recv(env, descP, opRef);
+ } else if (COMPARE(op, esock_atom_recvfrom) == 0) {
+ return ncancel_recv(env, descP, opRef);
+ } else if (COMPARE(op, esock_atom_recvmsg) == 0) {
+ return ncancel_recv(env, descP, opRef);
+ } else {
+ return esock_make_error(env, esock_atom_einval);
+ }
+}
+
+
+
+/* *** ncancel_connect ***
+ *
+ *
+ */
+static
+ERL_NIF_TERM ncancel_connect(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM opRef)
+{
+ return ncancel_write_select(env, descP, opRef);
+}
+
+
+/* *** ncancel_accept ***
+ *
+ * We have two different cases:
+ * *) Its the current acceptor
+ * Cancel the select!
+ * We need to activate one of the waiting acceptors.
+ * *) Its one of the acceptors ("waiting") in the queue
+ * Simply remove the acceptor from the queue.
+ *
+ */
+static
+ERL_NIF_TERM ncancel_accept(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM opRef)
+{
+ ERL_NIF_TERM res;
+
+ SSDBG( descP,
+ ("SOCKET", "ncancel_accept -> entry with"
+ "\r\n opRef: %T"
+ "\r\n %s"
+ "\r\n", opRef,
+ ((descP->currentAcceptorP == NULL) ? "without acceptor" : "with acceptor")) );
+
+ MLOCK(descP->accMtx);
+
+ if (descP->currentAcceptorP != NULL) {
+ if (COMPARE(opRef, descP->currentAcceptor.ref) == 0) {
+ res = ncancel_accept_current(env, descP);
+ } else {
+ res = ncancel_accept_waiting(env, descP, opRef);
+ }
+ } else {
+ /* Or badarg? */
+ res = esock_make_error(env, esock_atom_einval);
+ }
+
+ MUNLOCK(descP->accMtx);
+
+ SSDBG( descP,
+ ("SOCKET", "ncancel_accept -> done with result:"
+ "\r\n %T"
+ "\r\n", res) );
+
+ return res;
+}
+
+
+/* The current acceptor process has an ongoing select we first must
+ * cancel. Then we must re-activate the "first" (the first
+ * in the acceptor queue).
+ */
+static
+ERL_NIF_TERM ncancel_accept_current(ErlNifEnv* env,
+ SocketDescriptor* descP)
+{
+ ERL_NIF_TERM res;
+
+ SSDBG( descP, ("SOCKET", "ncancel_accept_current -> entry\r\n") );
+
+ res = ncancel_read_select(env, descP, descP->currentAcceptor.ref);
+
+ SSDBG( descP, ("SOCKET", "ncancel_accept_current -> cancel res: %T\r\n", res) );
+
+ if (acceptor_pop(env, descP,
+ &descP->currentAcceptor.pid,
+ &descP->currentAcceptor.mon,
+ &descP->currentAcceptor.ref)) {
+
+ /* There was another one */
+
+ SSDBG( descP, ("SOCKET", "ncancel_accept_current -> new (active) acceptor: "
+ "\r\n pid: %T"
+ "\r\n ref: %T"
+ "\r\n",
+ descP->currentAcceptor.pid,
+ descP->currentAcceptor.ref) );
+
+ SELECT(env,
+ descP->sock,
+ (ERL_NIF_SELECT_READ),
+ descP, &descP->currentAcceptor.pid, descP->currentAcceptor.ref);
+
+ } else {
+ SSDBG( descP, ("SOCKET", "ncancel_accept_current -> no more acceptors\r\n") );
+ descP->currentAcceptorP = NULL;
+ descP->state = SOCKET_STATE_LISTENING;
+ }
+
+ SSDBG( descP, ("SOCKET", "ncancel_accept_current -> done with result:"
+ "\r\n %T"
+ "\r\n", res) );
+
+ return res;
+}
+
+
+/* These processes have not performed a select, so we can simply
+ * remove them from the acceptor queue.
+ */
+static
+ERL_NIF_TERM ncancel_accept_waiting(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM opRef)
+{
+ ErlNifPid caller;
+
+ if (enif_self(env, &caller) == NULL)
+ return esock_make_error(env, atom_exself);
+
+ /* unqueue request from (acceptor) queue */
+
+ if (acceptor_unqueue(env, descP, &caller)) {
+ return esock_atom_ok;
+ } else {
+ /* Race? */
+ return esock_make_error(env, esock_atom_not_found);
+ }
+}
+
+
+
+/* *** ncancel_send ***
+ *
+ * Cancel a send operation.
+ * Its either the current writer or one of the waiting writers.
+ */
+static
+ERL_NIF_TERM ncancel_send(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM opRef)
+{
+ ERL_NIF_TERM res;
+
+ SSDBG( descP,
+ ("SOCKET", "ncancel_send -> entry with"
+ "\r\n opRef: %T"
+ "\r\n %s"
+ "\r\n", opRef,
+ ((descP->currentWriterP == NULL) ? "without writer" : "with writer")) );
+
+ MLOCK(descP->writeMtx);
+
+ if (descP->currentWriterP != NULL) {
+ if (COMPARE(opRef, descP->currentWriter.ref) == 0) {
+ res = ncancel_send_current(env, descP);
+ } else {
+ res = ncancel_send_waiting(env, descP, opRef);
+ }
+ } else {
+ /* Or badarg? */
+ res = esock_make_error(env, esock_atom_einval);
+ }
+
+ MUNLOCK(descP->writeMtx);
+
+ SSDBG( descP,
+ ("SOCKET", "ncancel_send -> done with result:"
+ "\r\n %T"
+ "\r\n", res) );
+
+ return res;
+}
+
+
+
+/* The current writer process has an ongoing select we first must
+ * cancel. Then we must re-activate the "first" (the first
+ * in the writer queue).
+ */
+static
+ERL_NIF_TERM ncancel_send_current(ErlNifEnv* env,
+ SocketDescriptor* descP)
+{
+ ERL_NIF_TERM res;
+
+ SSDBG( descP, ("SOCKET", "ncancel_send_current -> entry\r\n") );
+
+ res = ncancel_write_select(env, descP, descP->currentWriter.ref);
+
+ SSDBG( descP, ("SOCKET", "ncancel_send_current -> cancel res: %T\r\n", res) );
+
+ if (writer_pop(env, descP,
+ &descP->currentWriter.pid,
+ &descP->currentWriter.mon,
+ &descP->currentWriter.ref)) {
+
+ /* There was another one */
+
+ SSDBG( descP, ("SOCKET", "ncancel_send_current -> new (active) writer: "
+ "\r\n pid: %T"
+ "\r\n ref: %T"
+ "\r\n",
+ descP->currentWriter.pid,
+ descP->currentWriter.ref) );
+
+ SELECT(env,
+ descP->sock,
+ (ERL_NIF_SELECT_WRITE),
+ descP, &descP->currentWriter.pid, descP->currentWriter.ref);
+
+ } else {
+ SSDBG( descP, ("SOCKET", "ncancel_send_current -> no more writers\r\n") );
+ descP->currentWriterP = NULL;
+ }
+
+ SSDBG( descP, ("SOCKET", "ncancel_send_current -> done with result:"
+ "\r\n %T"
+ "\r\n", res) );
+
+ return res;
+}
+
+
+/* These processes have not performed a select, so we can simply
+ * remove them from the writer queue.
+ */
+static
+ERL_NIF_TERM ncancel_send_waiting(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM opRef)
+{
+ ErlNifPid caller;
+
+ if (enif_self(env, &caller) == NULL)
+ return esock_make_error(env, atom_exself);
+
+ /* unqueue request from (writer) queue */
+
+ if (writer_unqueue(env, descP, &caller)) {
+ return esock_atom_ok;
+ } else {
+ /* Race? */
+ return esock_make_error(env, esock_atom_not_found);
+ }
+}
+
+
+
+/* *** ncancel_recv ***
+ *
+ *
+ */
+static
+ERL_NIF_TERM ncancel_recv(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM opRef)
+{
+ return esock_make_error(env, esock_atom_einval);
+}
+
+
+
+static
+ERL_NIF_TERM ncancel_read_select(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM opRef)
+{
+ return ncancel_mode_select(env, descP, opRef,
+ ERL_NIF_SELECT_READ,
+ ERL_NIF_SELECT_READ_CANCELLED);
+}
+
+
+static
+ERL_NIF_TERM ncancel_write_select(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM opRef)
+{
+ return ncancel_mode_select(env, descP, opRef,
+ ERL_NIF_SELECT_WRITE,
+ ERL_NIF_SELECT_WRITE_CANCELLED);
+}
+
+
+static
+ERL_NIF_TERM ncancel_mode_select(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM opRef,
+ int smode,
+ int rmode)
+{
+ int selectRes = enif_select(env, descP->sock,
+ (ERL_NIF_SELECT_CANCEL | smode),
+ descP, NULL, opRef);
+
+ if (selectRes & rmode) {
+ /* Was cancelled */
+ return esock_atom_ok;
+ } else if (selectRes > 0) {
+ /* Has already sent the message */
+ return esock_make_error(env, esock_atom_select_sent);
+ } else {
+ /* Stopped? */
+ SSDBG( descP, ("SOCKET", "ncancel_mode_select -> failed: %d (0x%lX)"
+ "\r\n", selectRes, selectRes) );
+ return esock_make_error(env, esock_atom_einval);
+ }
+
+}
+
+
+
+/* ----------------------------------------------------------------------
* U t i l i t y F u n c t i o n s
* ----------------------------------------------------------------------
*/
+/* *** send_check_writer ***
+ *
+ * Checks if we have a current writer and if that is us. If not, then we must
+ * be made to wait for our turn. This is done by pushing us unto the writer queue.
+ */
+static
+BOOLEAN_T send_check_writer(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ERL_NIF_TERM ref,
+ ERL_NIF_TERM* checkResult)
+{
+ if (descP->currentWriterP != NULL) {
+ ErlNifPid caller;
+
+ if (enif_self(env, &caller) == NULL) {
+ *checkResult = esock_make_error(env, atom_exself);
+ return FALSE;
+ }
+
+ if (!compare_pids(env, &descP->currentWriter.pid, &caller)) {
+ /* Not the "current writer", so (maybe) push onto queue */
+
+ SSDBG( descP,
+ ("SOCKET", "send_check_writer -> not (current) writer\r\n") );
+
+ if (!writer_search4pid(env, descP, &caller))
+ *checkResult = writer_push(env, descP, caller, ref);
+ else
+ *checkResult = esock_make_error(env, esock_atom_eagain);
+
+ SSDBG( descP,
+ ("SOCKET",
+ "nsend -> queue (push) result: %T\r\n", checkResult) );
+
+ return FALSE;
+
+ }
+
+ }
+
+ *checkResult = esock_atom_ok; // Does not actually matter in this case, but ...
+
+ return TRUE;
+}
+
+
+
+/* *** send_check_result ***
+ *
+ * Check the result of a socket send (send, sendto and sendmsg) call.
+ * If a "complete" send has been made, the next (waiting) writer will be
+ * scheduled (if there is one).
+ * If we did not manage to send the entire package, make another select,
+ * so that we can be informed when we can make another try (to send the rest),
+ * and return with the amount we actually managed to send (its up to the caller
+ * (that is the erlang code) to figure out hust much is left to send).
+ * If the write fail, we give up and return with the appropriate error code.
+ *
+ * What about the remaining writers!!
+ */
static
ERL_NIF_TERM send_check_result(ErlNifEnv* env,
SocketDescriptor* descP,
@@ -10437,24 +11005,67 @@ ERL_NIF_TERM send_check_result(ErlNifEnv* env,
cnt_inc(&descP->writePkgCnt, 1);
cnt_inc(&descP->writeByteCnt, written);
+ DEMONP(env, descP, &descP->currentWriter.mon);
SSDBG( descP,
("SOCKET", "send_check_result -> "
"everything written (%d,%d) - done\r\n", dataSize, written) );
+ /* Ok, this write is done maybe activate the next (if any) */
+
+ if (writer_pop(env, descP,
+ &descP->currentWriter.pid,
+ &descP->currentWriter.mon,
+ &descP->currentWriter.ref)) {
+
+ /* There was another one */
+
+ SSDBG( descP, ("SOCKET", "send_check_result -> new (active) writer: "
+ "\r\n pid: %T"
+ "\r\n ref: %T"
+ "\r\n",
+ descP->currentWriter.pid,
+ descP->currentWriter.ref) );
+
+ SELECT(env,
+ descP->sock,
+ (ERL_NIF_SELECT_WRITE),
+ descP, &descP->currentWriter.pid, descP->currentWriter.ref);
+
+ } else {
+ descP->currentWriterP = NULL;
+ }
+
return esock_atom_ok;
} else if (written < 0) {
- /* Ouch, check what kind of failure */
+ /* Some kind of send failure - check what kind */
+
if ((saveErrno != EAGAIN) && (saveErrno != EINTR)) {
+ ErlNifPid pid;
+ ErlNifMonitor mon;
+ ERL_NIF_TERM ref, res;
+
+ /*
+ * An actual failure - we (and everyone waiting) give up
+ */
cnt_inc(&descP->writeFails, 1);
SSDBG( descP,
("SOCKET", "send_check_result -> error: %d\r\n", saveErrno) );
- return esock_make_error_errno(env, saveErrno);
+ res = esock_make_error_errno(env, saveErrno);
+
+ while (writer_pop(env, descP, &pid, &mon, &ref)) {
+ SSDBG( descP,
+ ("SOCKET", "send_check_result -> abort %T\r\n", pid) );
+ send_msg_nif_abort(env, ref, res, &pid);
+ DEMONP(env, descP, &mon);
+ }
+
+ return res;
} else {
@@ -10531,8 +11142,9 @@ ERL_NIF_TERM recv_check_result(ErlNifEnv* env,
}
/* There is a special case: If the provided 'to read' value is
- * zero (0). That means that we reads as much as we can, using
- * the default read buffer size.
+ * zero (0) (only for type =/= stream).
+ * That means that we reads as much as we can, using the default
+ * read buffer size.
*/
if (bufP->size == read) {
@@ -12541,6 +13153,8 @@ SocketDescriptor* alloc_descriptor(SOCKET sock, HANDLE event)
if ((descP = enif_alloc_resource(sockets, sizeof(SocketDescriptor))) != NULL) {
char buf[64]; /* Buffer used for building the mutex name */
+ descP->env = enif_alloc_env();
+
sprintf(buf, "socket[w,%d]", sock);
descP->writeMtx = MCREATE(buf);
descP->currentWriterP = NULL; // currentWriter not used
@@ -13187,7 +13801,7 @@ ERL_NIF_TERM acceptor_push(ErlNifEnv* env,
SocketRequestor* reqP = &e->data;
reqP->pid = pid;
- reqP->ref = ref;
+ reqP->ref = enif_make_copy(descP->env, ref);
if (MONP(env, descP, &pid, &reqP->mon) > 0) {
FREE(reqP);
@@ -13231,6 +13845,108 @@ BOOLEAN_T acceptor_pop(ErlNifEnv* env,
}
+/* *** acceptor unqueue ***
+ *
+ * Remove an acceptor from the acceptor queue.
+ */
+static
+BOOLEAN_T acceptor_unqueue(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ const ErlNifPid* pid)
+{
+ return qunqueue(env, &descP->acceptorsQ, pid);
+}
+
+
+
+/* *** writer search for pid ***
+ *
+ * Search for a pid in the writer queue.
+ */
+static
+BOOLEAN_T writer_search4pid(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ErlNifPid* pid)
+{
+ return qsearch4pid(env, &descP->writersQ, pid);
+}
+
+
+/* *** writer push ***
+ *
+ * Push an writer onto the writer queue.
+ * This happens when we already have atleast one current writer.
+ */
+static
+ERL_NIF_TERM writer_push(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ErlNifPid pid,
+ ERL_NIF_TERM ref)
+{
+ SocketRequestQueueElement* e = MALLOC(sizeof(SocketRequestQueueElement));
+ SocketRequestor* reqP = &e->data;
+
+ reqP->pid = pid;
+ reqP->ref = enif_make_copy(descP->env, ref);
+
+ if (MONP(env, descP, &pid, &reqP->mon) > 0) {
+ FREE(reqP);
+ return esock_make_error(env, atom_exmon);
+ }
+
+ qpush(&descP->writersQ, e);
+
+ // THIS IS OK => MAKES THE CALLER WAIT FOR ITS TURN
+ return esock_make_error(env, esock_atom_eagain);
+}
+
+
+/* *** writer pop ***
+ *
+ * Pop an writer from the writer queue.
+ */
+static
+BOOLEAN_T writer_pop(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ ErlNifPid* pid,
+ ErlNifMonitor* mon,
+ ERL_NIF_TERM* ref)
+{
+ SocketRequestQueueElement* e = qpop(&descP->writersQ);
+
+ if (e != NULL) {
+ *pid = e->data.pid;
+ *mon = e->data.mon;
+ *ref = e->data.ref; // At this point the ref has already been copied (env)
+ FREE(e);
+ return TRUE;
+ } else {
+ /* (acceptors) Queue was empty */
+ // *pid = NULL; we have no null value for pids
+ // *mon = NULL; we have no null value for monitors
+ *ref = esock_atom_undefined; // Just in case
+ return FALSE;
+ }
+
+}
+
+
+/* *** writer unqueue ***
+ *
+ * Remove an writer from the writer queue.
+ */
+static
+BOOLEAN_T writer_unqueue(ErlNifEnv* env,
+ SocketDescriptor* descP,
+ const ErlNifPid* pid)
+{
+ return qunqueue(env, &descP->writersQ, pid);
+}
+
+
+
+
+
static
BOOLEAN_T qsearch4pid(ErlNifEnv* env,
SocketRequestQueue* q,
@@ -13673,7 +14389,7 @@ void socket_down(ErlNifEnv* env,
"socket_down -> "
"not current acceptor - maybe a waiting acceptor\r\n") );
- qunqueue(env, &descP->acceptorsQ, pid);
+ acceptor_unqueue(env, descP, pid);
}
}
@@ -13723,7 +14439,7 @@ ErlNifFunc socket_funcs[] =
* is called after the connect *select* has "completed".
*/
{"nif_finalize_connection", 1, nif_finalize_connection, 0},
- // {"nif_cancel", 2, nif_cancel, 0},
+ {"nif_cancel", 3, nif_cancel, 0},
{"nif_finalize_close", 1, nif_finalize_close, ERL_NIF_DIRTY_JOB_IO_BOUND}
};
@@ -13843,8 +14559,10 @@ int on_load(ErlNifEnv* env, void** priv_data, ERL_NIF_TERM load_info)
atom_want = MKA(env, str_want);
/* Global atom(s) */
+ esock_atom_accept = MKA(env, "accept");
esock_atom_addr = MKA(env, "addr");
esock_atom_any = MKA(env, "any");
+ esock_atom_connect = MKA(env, "connect");
esock_atom_credentials = MKA(env, "credentials");
esock_atom_ctrl = MKA(env, "ctrl");
esock_atom_ctrunc = MKA(env, "ctrunc");
@@ -13869,6 +14587,7 @@ int on_load(ErlNifEnv* env, void** priv_data, ERL_NIF_TERM load_info)
esock_atom_loopback = MKA(env, "loopback");
esock_atom_lowdelay = MKA(env, "lowdelay");
esock_atom_mincost = MKA(env, "mincost");
+ esock_atom_not_found = MKA(env, "not_found");
esock_atom_ok = MKA(env, "ok");
esock_atom_oob = MKA(env, "oob");
esock_atom_origdstaddr = MKA(env, "origdstaddr");
@@ -13878,11 +14597,18 @@ int on_load(ErlNifEnv* env, void** priv_data, ERL_NIF_TERM load_info)
esock_atom_protocol = MKA(env, "protocol");
esock_atom_raw = MKA(env, "raw");
esock_atom_rdm = MKA(env, "rdm");
+ esock_atom_recv = MKA(env, "recv");
+ esock_atom_recvfrom = MKA(env, "recvfrom");
+ esock_atom_recvmsg = MKA(env, "recvmsg");
esock_atom_reliability = MKA(env, "reliability");
esock_atom_rights = MKA(env, "rights");
esock_atom_scope_id = MKA(env, "scope_id");
esock_atom_sctp = MKA(env, "sctp");
esock_atom_sec = MKA(env, "sec");
+ esock_atom_select_sent = MKA(env, "select_sent");
+ esock_atom_send = MKA(env, "send");
+ esock_atom_sendmsg = MKA(env, "sendmsg");
+ esock_atom_sendto = MKA(env, "sendto");
esock_atom_seqpacket = MKA(env, "seqpacket");
esock_atom_socket = MKA(env, "socket");
esock_atom_spec_dst = MKA(env, "spec_dst");
diff --git a/erts/preloaded/ebin/socket.beam b/erts/preloaded/ebin/socket.beam
index e6a33337ba..9e6d9f4709 100644
--- a/erts/preloaded/ebin/socket.beam
+++ b/erts/preloaded/ebin/socket.beam
Binary files differ
diff --git a/erts/preloaded/src/socket.erl b/erts/preloaded/src/socket.erl
index ad7a35694b..652054457f 100644
--- a/erts/preloaded/src/socket.erl
+++ b/erts/preloaded/src/socket.erl
@@ -1137,7 +1137,7 @@ connect(#socket{ref = SockRef}, #{family := Fam} = SockAddr, Timeout)
%% </KOLLA>
nif_finalize_connection(SockRef)
after NewTimeout ->
- nif_cancel(SockRef, connect, Ref),
+ cancel(SockRef, connect, Ref),
{error, timeout}
end;
{error, _} = ERROR ->
@@ -1145,7 +1145,6 @@ connect(#socket{ref = SockRef}, #{family := Fam} = SockAddr, Timeout)
end.
-
%% ===========================================================================
%%
%% listen - listen for connections on a socket
@@ -1227,13 +1226,12 @@ do_accept(LSockRef, Timeout) ->
{error, Reason}
after NewTimeout ->
- nif_cancel(LSockRef, accept, AccRef),
- flush_select_msgs(LSockRef, AccRef),
+ cancel(LSockRef, accept, AccRef),
{error, timeout}
end;
{error, _} = ERROR ->
- nif_cancel(LSockRef, accept, AccRef), % Just to be on the safe side...
+ cancel(LSockRef, accept, AccRef), % Just to be on the safe side...
ERROR
end.
@@ -1305,8 +1303,7 @@ do_send(SockRef, Data, EFlags, Timeout) ->
{error, Reason}
after NewTimeout ->
- nif_cancel(SockRef, send, SendRef),
- flush_select_msgs(SockRef, SendRef),
+ cancel(SockRef, send, SendRef),
{error, {timeout, size(Data)}}
end;
{error, eagain} ->
@@ -1319,8 +1316,7 @@ do_send(SockRef, Data, EFlags, Timeout) ->
{error, Reason}
after Timeout ->
- nif_cancel(SockRef, send, SendRef),
- flush_select_msgs(SockRef, SendRef),
+ cancel(SockRef, send, SendRef),
{error, {timeout, size(Data)}}
end;
@@ -1349,10 +1345,19 @@ sendto(Socket, Data, Dest) ->
Data :: binary(),
Dest :: null | sockaddr(),
Flags :: send_flags(),
- Reason :: term().
+ Reason :: term()
+ ; (Socket, Data, Dest, Timeout) -> ok | {error, Reason} when
+ Socket :: socket(),
+ Data :: iodata(),
+ Dest :: null | sockaddr(),
+ Timeout :: timeout(),
+ Reason :: term().
+
+sendto(Socket, Data, Dest, Flags) when is_list(Flags) ->
+ sendto(Socket, Data, Dest, Flags, ?SOCKET_SENDTO_TIMEOUT_DEFAULT);
+sendto(Socket, Data, Dest, Timeout) ->
+ sendto(Socket, Data, Dest, ?SOCKET_SENDTO_FLAGS_DEFAULT, Timeout).
-sendto(Socket, Data, Dest, Flags) ->
- sendto(Socket, Data, Dest, Flags, ?SOCKET_SENDTO_TIMEOUT_DEFAULT).
-spec sendto(Socket, Data, Dest, Flags, Timeout) -> ok | {error, Reason} when
Socket :: socket(),
@@ -1403,8 +1408,7 @@ do_sendto(SockRef, Data, Dest, EFlags, Timeout) ->
{error, Reason}
after Timeout ->
- nif_cancel(SockRef, sendto, SendRef),
- flush_select_msgs(SockRef, SendRef),
+ cancel(SockRef, sendto, SendRef),
{error, timeout}
end;
@@ -1414,8 +1418,7 @@ do_sendto(SockRef, Data, Dest, EFlags, Timeout) ->
do_sendto(SockRef, Data, Dest, EFlags,
next_timeout(TS, Timeout))
after Timeout ->
- nif_cancel(SockRef, sendto, SendRef),
- flush_select_msgs(SockRef, SendRef),
+ cancel(SockRef, sendto, SendRef),
{error, timeout}
end;
@@ -1497,8 +1500,7 @@ do_sendmsg(SockRef, MsgHdr, EFlags, Timeout) ->
do_sendmsg(SockRef, MsgHdr, EFlags,
next_timeout(TS, Timeout))
after Timeout ->
- nif_cancel(SockRef, sendmsg, SendRef),
- flush_select_msgs(SockRef, SendRef),
+ cancel(SockRef, sendmsg, SendRef),
{error, timeout}
end;
@@ -1519,62 +1521,6 @@ ensure_msghdr(_) ->
%% writev - write data into multiple buffers
%%
-%% send(Socket, Data, Flags, Timeout)
-%% when (is_list(Data) orelse is_binary(Data)) andalso is_list(Flags) ->
-%% IOVec = erlang:iolist_to_iovec(Data),
-%% EFlags = enc_send_flags(Flags),
-%% send_iovec(Socket, IOVec, EFlags, Timeout).
-
-
-%% %% Iterate over the IO-vector (list of binaries).
-
-%% send_iovec(_Socket, [] = _IOVec, _EFlags, _Timeout) ->
-%% ok;
-%% send_iovec({socket, _, SockRef} = Socket, [Bin|IOVec], EFlags, Timeout) ->
-%% case do_send(SockRef, make_ref(), Bin, EFlags, Timeout) of
-%% {ok, NewTimeout} ->
-%% send_iovec(Socket, IOVec, EFlags, NewTimeout);
-%% {error, _} = ERROR ->
-%% ERROR
-%% end.
-
-
-%% do_send(SockRef, SendRef, Data, _EFlags, Timeout)
-%% when (Timeout < 0) ->
-%% nif_cancel(SockRef, SendRef),
-%% flush_select_msgs(SockRef, SendRef),
-%% {error, {timeout, size(Data)}};
-%% do_send(SockRef, SendRef, Data, EFlags, Timeout) ->
-%% TS = timestamp(Timeout),
-%% case nif_send(SockRef, SendRef, Data, EFlags) of
-%% ok ->
-%% {ok, next_timeout(TS, Timeout)};
-%% {ok, Written} ->
-%% %% We are partially done, wait for continuation
-%% receive
-%% {select, SockRef, SendRef, ready_output} ->
-%% <<_:Written/binary, Rest/binary>> = Data,
-%% do_send(SockRef, make_ref(), Rest, EFlags,
-%% next_timeout(TS, Timeout))
-%% after Timeout ->
-%% nif_cancel(SockRef, SendRef),
-%% flush_select_msgs(SockRef, SendRef),
-%% {error, timeout}
-%% end;
-%% {error, eagain} ->
-%% receive
-%% {select, SockRef, SendRef, ready_output} ->
-%% do_send(SockRef, SendRef, Data, EFlags,
-%% next_timeout(TS, Timeout))
-%% after Timeout ->
-%% nif_cancel(SockRef, SendRef),
-%% flush_select_msgs(SockRef, SendRef),
-%% {error, timeout}
-%% end;
-
-%% {error, _} = ERROR ->
-%% ERROR
-%% end.
%% ===========================================================================
@@ -1695,8 +1641,7 @@ do_recv(SockRef, _OldRef, Length, EFlags, Acc, Timeout)
after NewTimeout ->
- nif_cancel(SockRef, recv, RecvRef),
- flush_select_msgs(SockRef, RecvRef),
+ cancel(SockRef, recv, RecvRef),
{error, {timeout, Acc}}
end;
@@ -1715,8 +1660,7 @@ do_recv(SockRef, _OldRef, Length, EFlags, Acc, Timeout)
after NewTimeout ->
- nif_cancel(SockRef, recv, RecvRef),
- flush_select_msgs(SockRef, RecvRef),
+ cancel(SockRef, recv, RecvRef),
{error, {timeout, Acc}}
end;
@@ -1739,8 +1683,7 @@ do_recv(SockRef, _OldRef, Length, EFlags, Acc, Timeout)
{error, Reason}
after NewTimeout ->
- nif_cancel(SockRef, recv, RecvRef),
- flush_select_msgs(SockRef, RecvRef),
+ cancel(SockRef, recv, RecvRef),
{error, timeout}
end;
@@ -1765,7 +1708,7 @@ do_recv(SockRef, RecvRef, 0 = _Length, _Eflags, Acc, _Timeout) ->
%% The current recv operation is to be cancelled, so no need for a ref...
%% The cancel will end our 'read everything you have' and "activate"
%% any waiting reader.
- nif_cancel(SockRef, recv, RecvRef),
+ cancel(SockRef, recv, RecvRef),
{ok, Acc};
do_recv(_SockRef, _RecvRef, _Length, _EFlags, Acc, _Timeout) when (size(Acc) > 0) ->
{error, {timeout, Acc}};
@@ -1878,8 +1821,7 @@ do_recvfrom(SockRef, BufSz, EFlags, Timeout) ->
{error, Reason}
after NewTimeout ->
- nif_cancel(SockRef, recvfrom, RecvRef),
- flush_select_msgs(SockRef, RecvRef),
+ cancel(SockRef, recvfrom, RecvRef),
{error, timeout}
end;
@@ -1966,8 +1908,7 @@ do_recvmsg(SockRef, BufSz, CtrlSz, EFlags, Timeout) ->
{error, Reason}
after NewTimeout ->
- nif_cancel(SockRef, recvmsg, RecvRef),
- flush_select_msgs(SockRef, RecvRef),
+ cancel(SockRef, recvmsg, RecvRef),
{error, timeout}
end;
@@ -3325,10 +3266,19 @@ ensure_sockaddr(_SockAddr) ->
-flush_select_msgs(LSRef, Ref) ->
+cancel(SockRef, Op, OpRef) ->
+ case nif_cancel(SockRef, Op, OpRef) of
+ %% The select has already completed
+ {error, select_sent} ->
+ flush_select_msgs(SockRef, OpRef);
+ Other ->
+ Other
+ end.
+
+flush_select_msgs(SockRef, Ref) ->
receive
- {select, LSRef, Ref, _} ->
- flush_select_msgs(LSRef, Ref)
+ {select, SockRef, Ref, _} ->
+ flush_select_msgs(SockRef, Ref)
after 0 ->
ok
end.
generated by cgit v1.2.3 (git 2.25.1) at 2024-09-01 12:17:31 +0000